<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.jackrabbit.core.data;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.lang.ref.WeakReference;
import java.security.DigestOutputStream;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.WeakHashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import javax.jcr.RepositoryException;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.jackrabbit.core.data.util.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A caching data store that consists of {@link LocalCache} and {@link Backend}.
 * {@link Backend} is single source of truth. All methods first try to fetch
 * information from {@link LocalCache}. If record is not available in
 * {@link LocalCache}, then it is fetched from {@link Backend} and saved to
 * {@link LocalCache} for further access. This class is designed to work without
 * {@link LocalCache} and then all information is fetched from {@link Backend}.
 * To disable {@link LocalCache} set {@link #setCacheSize(long)} to 0. *
 * Configuration:
 * 
 * <pre>
 * &lt;DataStore class="org.apache.jackrabbit.aws.ext.ds.CachingDataStore">
 * 
 *     &lt;param name="{@link #setPath(String) path}" value="/data/datastore"/>
 *     &lt;param name="{@link #setConfig(String) config}" value="${rep.home}/backend.properties"/>
 *     &lt;param name="{@link #setCacheSize(long) cacheSize}" value="68719476736"/>
 *     &lt;param name="{@link #setSecret(String) secret}" value="123456"/>
 *     &lt;param name="{@link #setCachePurgeTrigFactor(double)}" value="0.95d"/>
 *     &lt;param name="{@link #setCachePurgeResizeFactor(double) cacheSize}" value="0.85d"/>
 *     &lt;param name="{@link #setMinRecordLength(int) minRecordLength}" value="1024"/>
 *     &lt;param name="{@link #setContinueOnAsyncUploadFailure(boolean) continueOnAsyncUploadFailure}" value="false"/>
 *     &lt;param name="{@link #setConcurrentUploadsThreads(int) concurrentUploadsThreads}" value="10"/>
 *     &lt;param name="{@link #setAsyncUploadLimit(int) asyncUploadLimit}" value="100"/>
 *     &lt;param name="{@link #setUploadRetries(int) uploadRetries}" value="3"/>
 * &lt/DataStore>
 */
public abstract class CachingDataStore extends AbstractDataStore implements
        MultiDataStoreAware, AsyncUploadCallback {

    /**
     * Logger instance.
     */
    private static final Logger LOG = LoggerFactory.getLogger(CachingDataStore.class);

    /**
     * The digest algorithm used to uniquely identify records.
     */
    private static final String DIGEST = "SHA-1";

    private static final String DS_STORE = ".DS_Store";

    /**
     * Name of the directory used for temporary files. Must be at least 3
     * characters.
     */
    private static final String TMP = "tmp";

    /**
     * All data identifiers that are currently in use are in this set until they
     * are garbage collected.
     */
    protected Map<DataIdentifier, WeakReference<DataIdentifier>> inUse = Collections.synchronizedMap(new WeakHashMap<DataIdentifier, WeakReference<DataIdentifier>>());
    
    /**
     * In memory map to hold failed asynchronous upload {@link DataIdentifier}
     * and its retry count. Once if all retries are exhausted or file is 
     * successfully uploaded, then corresponding entry is flushed from the map.
     * As all failed uploads are synchronously uploaded at startup, this map 
     * is not required to be persisted. 
     */
    protected final Map<DataIdentifier, Integer> uploadRetryMap = new ConcurrentHashMap<DataIdentifier, Integer>(5);

    protected Backend backend;

    /**
     * The minimum size of an object that should be stored in this data store.
     */
    private int minRecordLength = 16 * 1024;

    private String path;

    private File directory;

    private File tmpDir;

    private String secret;

    /**
     * The optional backend configuration.
     */
    private String config;

    /**
     * The minimum modified date. If a file is accessed (read or write) with a
     * modified date older than this value, the modified date is updated to the
     * current time.
     */
    private long minModifiedDate;

    /**
     * Cache purge trigger factor. Cache will undergo in auto-purge mode if
     * cache current size is greater than cachePurgeTrigFactor * cacheSize
     */
    private double cachePurgeTrigFactor = 0.95d;

    /**
     * Cache resize factor. After auto-purge mode, cache current size would just
     * greater than cachePurgeResizeFactor * cacheSize cacheSize
     */
    private double cachePurgeResizeFactor = 0.85d;

    /**
     * The number of bytes in the cache. The default value is 64 GB.
     */
    private long cacheSize = 64L * 1024 * 1024 * 1024;
    
    /**
     * The number of retries for failed upload.
     */
    private int uploadRetries = 3;

    /**
     * The local file system cache.
     */
    private LocalCache cache;

    /**
     * Caching holding pending uploads
     */
    private AsyncUploadCache asyncWriteCache;

    protected abstract Backend createBackend();

    protected abstract String getMarkerFile();

    /**
     * In {@link #init(String)},it resumes all incomplete asynchronous upload
     * from {@link AsyncUploadCache} and uploads them concurrently in multiple
     * threads. It throws {@link RepositoryException}, if file is not found in
     * local cache for that asynchronous upload. As far as code is concerned, it
     * is only possible when somebody has removed files from local cache
     * manually. If there is an exception and user want to proceed with
     * inconsistencies, set parameter continueOnAsyncUploadFailure to true in
     * repository.xml. This will ignore {@link RepositoryException} and log all
     * missing files and proceed after resetting {@link AsyncUploadCache} .
     */
    private boolean continueOnAsyncUploadFailure;

    /**
     * The {@link #init(String)} methods checks for {@link #getMarkerFile()} and
     * if it doesn't exists migrates all files from fileystem to {@link Backend}
     * . This parameter governs number of threads which will upload files
     * concurrently to {@link Backend}.
     */
    private int concurrentUploadsThreads = 10;

    /**
     * This parameter limits the number of asynchronous uploads slots to
     * {@link Backend}. Once this limit is reached, further uploads to
     * {@link Backend} are synchronous, till one of asynchronous uploads
     * completes and make asynchronous uploads slot available. To disable
     * asynchronous upload, set {@link #asyncUploadLimit} parameter to 0 in
     * repository.xml. By default it is 100
     */
    private int asyncUploadLimit = 100;

    /**
     * Initialized the data store. If the path is not set, &lt;repository
     * home&gt;/repository/datastore is used. This directory is automatically
     * created if it does not yet exist. During first initialization, it upload
     * all files from local datastore to backed and local datastore act as a
     * local cache.
     */
    @Override
    public void init(String homeDir) throws RepositoryException {
        try {
            if (path == null) {
                path = homeDir + "/repository/datastore";
                tmpDir = new File(homeDir, "/repository/s3tmp");
            } else {
                // cache is moved from 'path' to 'path'/repository/datastore
                tmpDir = new File(path, "/repository/s3tmp");
                path = path + "/repository/datastore";
            }
            LOG.info("path=[{}],  tmpPath=[{}]", path, tmpDir.getPath());
            directory = new File(path);
            mkdirs(directory);
            if (!mkdirs(tmpDir)) {
                FileUtils.cleanDirectory(tmpDir);
                LOG.info("tmp=[{}] cleaned.", tmpDir.getPath());
            }

            asyncWriteCache = new AsyncUploadCache();
            asyncWriteCache.init(homeDir, path, asyncUploadLimit);

            backend = createBackend();
            backend.init(this, path, config);
            String markerFileName = getMarkerFile();
            if (markerFileName != null) {
                // create marker file in homeDir to avoid deletion in cache
                // cleanup.
                File markerFile = new File(homeDir, markerFileName);
                if (!markerFile.exists()) {
                    LOG.info("load files from local cache");
                    uploadFilesFromCache();
                    try {
                        markerFile.createNewFile();
                    } catch (IOException e) {
                        throw new DataStoreException(
                            "Could not create marker file "
                                + markerFile.getAbsolutePath(), e);
                    }
                } else {
                    LOG.info("marker file = [{}] exists ",
                        markerFile.getAbsolutePath());
                }
            }
            // upload any leftover async uploads to backend during last shutdown
            Set<String> fileList = asyncWriteCache.getAll();
            if (fileList != null && !fileList.isEmpty()) {
                List<String> errorFiles = new ArrayList<String>();
                LOG.info("Uploading [{}] and size=[{}] from AsyncUploadCache.",
                    fileList, fileList.size());
                long totalSize = 0;
                List<File> files = new ArrayList<File>(fileList.size());
                for (String fileName : fileList) {
                    File f = new File(path, fileName);
                    if (!f.exists()) {
                        errorFiles.add(fileName);
                        LOG.error(
                            "Cannot upload pending file [{}]. File doesn't exist.",
                            f.getAbsolutePath());
                    } else {
                        totalSize += f.length();
                        files.add(new File(homeDir, fileName));
                    }
                }
                new FilesUploader(files, totalSize, concurrentUploadsThreads,
                    true).upload();
                if (!continueOnAsyncUploadFailure && errorFiles.size() > 0) {
                    LOG.error(
                        "Pending uploads of files [{}] failed. Files do not exist in Local cache.",
                        errorFiles);
                    LOG.error("To continue set [continueOnAsyncUploadFailure] "
                        + "to true in Datastore configuration in "
                        + "repository.xml. There would be inconsistent data "
                        + "in repository due the missing files. ");
                    throw new RepositoryException(
                        "Cannot upload async uploads from local cache. Files not found.");
                } else {
                    if (errorFiles.size() > 0) {
                        LOG.error(
                            "Pending uploads of files [{}] failed. Files do" +
                            " not exist in Local cache. Continuing as " +<commit_msg>JCR-3771 - Pending async uploads fails to get uploaded on restart. <commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.jackrabbit.core.data;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.lang.ref.WeakReference;
import java.security.DigestOutputStream;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.WeakHashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import javax.jcr.RepositoryException;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.jackrabbit.core.data.util.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A caching data store that consists of {@link LocalCache} and {@link Backend}.
 * {@link Backend} is single source of truth. All methods first try to fetch
 * information from {@link LocalCache}. If record is not available in
 * {@link LocalCache}, then it is fetched from {@link Backend} and saved to
 * {@link LocalCache} for further access. This class is designed to work without
 * {@link LocalCache} and then all information is fetched from {@link Backend}.
 * To disable {@link LocalCache} set {@link #setCacheSize(long)} to 0. *
 * Configuration:
 * 
 * <pre>
 * &lt;DataStore class="org.apache.jackrabbit.aws.ext.ds.CachingDataStore">
 * 
 *     &lt;param name="{@link #setPath(String) path}" value="/data/datastore"/>
 *     &lt;param name="{@link #setConfig(String) config}" value="${rep.home}/backend.properties"/>
 *     &lt;param name="{@link #setCacheSize(long) cacheSize}" value="68719476736"/>
 *     &lt;param name="{@link #setSecret(String) secret}" value="123456"/>
 *     &lt;param name="{@link #setCachePurgeTrigFactor(double)}" value="0.95d"/>
 *     &lt;param name="{@link #setCachePurgeResizeFactor(double) cacheSize}" value="0.85d"/>
 *     &lt;param name="{@link #setMinRecordLength(int) minRecordLength}" value="1024"/>
 *     &lt;param name="{@link #setContinueOnAsyncUploadFailure(boolean) continueOnAsyncUploadFailure}" value="false"/>
 *     &lt;param name="{@link #setConcurrentUploadsThreads(int) concurrentUploadsThreads}" value="10"/>
 *     &lt;param name="{@link #setAsyncUploadLimit(int) asyncUploadLimit}" value="100"/>
 *     &lt;param name="{@link #setUploadRetries(int) uploadRetries}" value="3"/>
 * &lt/DataStore>
 */
public abstract class CachingDataStore extends AbstractDataStore implements
        MultiDataStoreAware, AsyncUploadCallback {

    /**
     * Logger instance.
     */
    private static final Logger LOG = LoggerFactory.getLogger(CachingDataStore.class);

    /**
     * The digest algorithm used to uniquely identify records.
     */
    private static final String DIGEST = "SHA-1";

    private static final String DS_STORE = ".DS_Store";

    /**
     * Name of the directory used for temporary files. Must be at least 3
     * characters.
     */
    private static final String TMP = "tmp";

    /**
     * All data identifiers that are currently in use are in this set until they
     * are garbage collected.
     */
    protected Map<DataIdentifier, WeakReference<DataIdentifier>> inUse = Collections.synchronizedMap(new WeakHashMap<DataIdentifier, WeakReference<DataIdentifier>>());
    
    /**
     * In memory map to hold failed asynchronous upload {@link DataIdentifier}
     * and its retry count. Once if all retries are exhausted or file is 
     * successfully uploaded, then corresponding entry is flushed from the map.
     * As all failed uploads are synchronously uploaded at startup, this map 
     * is not required to be persisted. 
     */
    protected final Map<DataIdentifier, Integer> uploadRetryMap = new ConcurrentHashMap<DataIdentifier, Integer>(5);

    protected Backend backend;

    /**
     * The minimum size of an object that should be stored in this data store.
     */
    private int minRecordLength = 16 * 1024;

    private String path;

    private File directory;

    private File tmpDir;

    private String secret;

    /**
     * The optional backend configuration.
     */
    private String config;

    /**
     * The minimum modified date. If a file is accessed (read or write) with a
     * modified date older than this value, the modified date is updated to the
     * current time.
     */
    private long minModifiedDate;

    /**
     * Cache purge trigger factor. Cache will undergo in auto-purge mode if
     * cache current size is greater than cachePurgeTrigFactor * cacheSize
     */
    private double cachePurgeTrigFactor = 0.95d;

    /**
     * Cache resize factor. After auto-purge mode, cache current size would just
     * greater than cachePurgeResizeFactor * cacheSize cacheSize
     */
    private double cachePurgeResizeFactor = 0.85d;

    /**
     * The number of bytes in the cache. The default value is 64 GB.
     */
    private long cacheSize = 64L * 1024 * 1024 * 1024;
    
    /**
     * The number of retries for failed upload.
     */
    private int uploadRetries = 3;

    /**
     * The local file system cache.
     */
    private LocalCache cache;

    /**
     * Caching holding pending uploads
     */
    private AsyncUploadCache asyncWriteCache;

    protected abstract Backend createBackend();

    protected abstract String getMarkerFile();

    /**
     * In {@link #init(String)},it resumes all incomplete asynchronous upload
     * from {@link AsyncUploadCache} and uploads them concurrently in multiple
     * threads. It throws {@link RepositoryException}, if file is not found in
     * local cache for that asynchronous upload. As far as code is concerned, it
     * is only possible when somebody has removed files from local cache
     * manually. If there is an exception and user want to proceed with
     * inconsistencies, set parameter continueOnAsyncUploadFailure to true in
     * repository.xml. This will ignore {@link RepositoryException} and log all
     * missing files and proceed after resetting {@link AsyncUploadCache} .
     */
    private boolean continueOnAsyncUploadFailure;

    /**
     * The {@link #init(String)} methods checks for {@link #getMarkerFile()} and
     * if it doesn't exists migrates all files from fileystem to {@link Backend}
     * . This parameter governs number of threads which will upload files
     * concurrently to {@link Backend}.
     */
    private int concurrentUploadsThreads = 10;

    /**
     * This parameter limits the number of asynchronous uploads slots to
     * {@link Backend}. Once this limit is reached, further uploads to
     * {@link Backend} are synchronous, till one of asynchronous uploads
     * completes and make asynchronous uploads slot available. To disable
     * asynchronous upload, set {@link #asyncUploadLimit} parameter to 0 in
     * repository.xml. By default it is 100
     */
    private int asyncUploadLimit = 100;

    /**
     * Initialized the data store. If the path is not set, &lt;repository
     * home&gt;/repository/datastore is used. This directory is automatically
     * created if it does not yet exist. During first initialization, it upload
     * all files from local datastore to backed and local datastore act as a
     * local cache.
     */
    @Override
    public void init(String homeDir) throws RepositoryException {
        try {
            if (path == null) {
                path = homeDir + "/repository/datastore";
                tmpDir = new File(homeDir, "/repository/s3tmp");
            } else {
                // cache is moved from 'path' to 'path'/repository/datastore
                tmpDir = new File(path, "/repository/s3tmp");
                path = path + "/repository/datastore";
            }
            LOG.info("path=[{}],  tmpPath=[{}]", path, tmpDir.getPath());
            directory = new File(path);
            mkdirs(directory);
            if (!mkdirs(tmpDir)) {
                FileUtils.cleanDirectory(tmpDir);
                LOG.info("tmp=[{}] cleaned.", tmpDir.getPath());
            }

            asyncWriteCache = new AsyncUploadCache();
            asyncWriteCache.init(homeDir, path, asyncUploadLimit);

            backend = createBackend();
            backend.init(this, path, config);
            String markerFileName = getMarkerFile();
            if (markerFileName != null) {
                // create marker file in homeDir to avoid deletion in cache
                // cleanup.
                File markerFile = new File(homeDir, markerFileName);
                if (!markerFile.exists()) {
                    LOG.info("load files from local cache");
                    uploadFilesFromCache();
                    try {
                        markerFile.createNewFile();
                    } catch (IOException e) {
                        throw new DataStoreException(
                            "Could not create marker file "
                                + markerFile.getAbsolutePath(), e);
                    }
                } else {
                    LOG.info("marker file = [{}] exists ",
                        markerFile.getAbsolutePath());
                }
            }
            // upload any leftover async uploads to backend during last shutdown
            Set<String> fileList = asyncWriteCache.getAll();
            if (fileList != null && !fileList.isEmpty()) {
                List<String> errorFiles = new ArrayList<String>();
                LOG.info("Uploading [{}] and size=[{}] from AsyncUploadCache.",
                    fileList, fileList.size());
                long totalSize = 0;
                List<File> files = new ArrayList<File>(fileList.size());
                for (String fileName : fileList) {
                    File f = new File(path, fileName);
                    if (!f.exists()) {
                        errorFiles.add(fileName);
                        LOG.error(
                            "Cannot upload pending file [{}]. File doesn't exist.",
                            f.getAbsolutePath());
                    } else {
                        totalSize += f.length();
                        files.add(new File(path, fileName));
                    }
                }
                new FilesUploader(files, totalSize, concurrentUploadsThreads,
                    true).upload();
                if (!continueOnAsyncUploadFailure && errorFiles.size() > 0) {
                    LOG.error(
                        "Pending uploads of files [{}] failed. Files do not exist in Local cache.",
                        errorFiles);
                    LOG.error("To continue set [continueOnAsyncUploadFailure] "
                        + "to true in Datastore configuration in "
                        + "repository.xml. There would be inconsistent data "
                        + "in repository due the missing files. ");
                    throw new RepositoryException(
                        "Cannot upload async uploads from local cache. Files not found.");
                } else {
                    if (errorFiles.size() > 0) {
                        LOG.error(
                            "Pending uploads of files [{}] failed. Files do" +
                            " not exist in Local cache. Continuing as " +
{'fraction_non_alphanumeric': 0.09382096330823474, 'fraction_numerical': 0.007011672841719454, 'mean_word_length': 2.417562968005446, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5221004', 'n_tokens_mistral': 6578, 'n_tokens_neox': 6174, 'n_words': 2220}