private RefStreamPurgeCounts purgeRefStreamDef(final TaskContext taskContext,
                                                   final RefStreamDefinition refStreamDefinition,
                                                   final RefDataProcessingInfo refDataProcessingInfo,
                                                   final ByteBuffer refStreamDefinitionBuf,
                                                   final BatchingWriteTxn batchingWriteTxn) {
        updatePurgeTaskContextInfoSupplier(
                taskContext, refStreamDefinition, "Starting purge", null);

        final Duration age = Duration.between(refDataProcessingInfo.getLastAccessedTime(), Instant.now());
        final String refStreamDefStr = LogUtil.message(
                "stream: {}, " +
                "state: {}, " +
                "access time: {} (age: {}), " +
                "store: '{}'" +
                "create time: {}, " +
                "effective time: {}, " +
                "pipeline: {}, " +
                "pipeline version: {}",
                refStreamDefinition.getStreamId(),
                refDataProcessingInfo.getProcessingState(),
                refDataProcessingInfo.getLastAccessedTime(),
                age,
                storeName,
                refDataProcessingInfo.getCreateTime(),
                refDataProcessingInfo.getEffectiveTime(),
                DocRefUtil.createSimpleDocRefString(refStreamDefinition.getPipelineDocRef()),
                refStreamDefinition.getPipelineVersion());

        LOGGER.info("Purging refStreamDefinition with {}", refStreamDefStr);

        RefStreamPurgeCounts refStreamSummaryInfo = RefStreamPurgeCounts.zero();
        final Instant refStreamStartTime = Instant.now();
        try {
            // mark it is purge in progress, so if we are committing part way through
            // other processes will know it is a partial state.
            processingInfoDb.updateProcessingState(batchingWriteTxn.getTxn(),
                    refStreamDefinitionBuf,
                    ProcessingState.PURGE_IN_PROGRESS,
                    false);

            // purge the data associated with this ref stream def
            refStreamSummaryInfo = purgeRefStreamData(taskContext, batchingWriteTxn, refStreamDefinition);

            //now delete the proc info entry
            LOGGER.debug("Deleting processing info entry for {}", refStreamDefinition);
            updatePurgeTaskContextInfoSupplier(
                    taskContext, refStreamDefinition, "Deleting processing info entry", null);

            boolean didDeleteSucceed = processingInfoDb.delete(
                    batchingWriteTxn.getTxn(), refStreamDefinitionBuf);

            if (!didDeleteSucceed) {
                throw new RuntimeException("Processing info entry not found so was not deleted");
            }

            // Ensure we commit at the end of each ref stream
            batchingWriteTxn.commit();

            LOGGER.info("Completed purge of refStreamDefinition with stream {} (" +
                        "{} maps deleted, {} values deleted, {} values de-referenced) in {}",
                    refStreamDefinition.getStreamId(),
                    refStreamSummaryInfo.mapsDeletedCount,
                    refStreamSummaryInfo.valuesDeletedCount,
                    refStreamSummaryInfo.valuesDeReferencedCount,
                    Duration.between(refStreamStartTime, Instant.now()));

        } catch (Exception e) {
            refStreamSummaryInfo = refStreamSummaryInfo.increment(
                    0, 0, 0, false);

            LOGGER.error("Failed purge of refStreamDefinition with stream {} (" +
                         "{} maps deleted, {} values deleted, {} values de-referenced) in {}: {}",
                    refStreamDefinition.getStreamId(),
                    refStreamSummaryInfo.mapsDeletedCount,
                    refStreamSummaryInfo.valuesDeletedCount,
                    refStreamSummaryInfo.valuesDeReferencedCount,
                    Duration.between(refStreamStartTime, Instant.now()),
                    e.getMessage(), e);

            try {
                // We are still under the ref stream def lock here
                processingInfoDb.updateProcessingState(
                        batchingWriteTxn.getTxn(),
                        refStreamDefinitionBuf,
                        ProcessingState.PURGE_FAILED,
                        false);
            } catch (Exception e2) {
                LOGGER.error("Unable to update processing state for ref stream {}: {}",
                        refStreamDefinition, e2.getMessage(), e2);
            }
            // Don't re-throw so we can move on to the next one
        }

        return refStreamSummaryInfo;
    }
Synchronized to prevent to ensure concurrent purge jobs don't clash

@param now Allows the setting of the current time for testing purposes
stroom-pipeline/src/main/java/stroom/pipeline/refdata/store/offheapstore/RefDataOffHeapStore.java