synchronized void purgeOldData(final Instant now, final StroomDuration purgeAge) {
        final TaskContext taskContext = taskContextFactory.current();
        taskContext.info(() -> "Purging old data in store " + storeName);
        final Instant startTime = Instant.now();
        final AtomicReference<PurgeCounts> countsRef = new AtomicReference<>(PurgeCounts.zero());
        final Instant purgeCutOffTime = TimeUtils.durationToThreshold(now, purgeAge);

        LOGGER.info("Checking ref store '{}' for streams to purge, with purge age: {} ({}), cut off time: {}",
                storeName,
                purgeAge,
                purgeAge.getDuration(),
                purgeCutOffTime);

        try (final PooledByteBuffer accessTimeThresholdPooledBuf = getAccessTimeCutOffBuffer(purgeCutOffTime);
                final PooledByteBufferPair procInfoPooledBufferPair = processingInfoDb.getPooledBufferPair()) {

            // Reference is initially empty so we will scan from the beginning of the DB
            final AtomicReference<ByteBuffer> currRefStreamDefBufRef = new AtomicReference<>();
            final ByteBuffer accessTimeThresholdBuf = accessTimeThresholdPooledBuf.getByteBuffer();

            final Predicate<ByteBuffer> accessTimePredicate = processingInfoBuffer ->
                    !RefDataProcessingInfoSerde.wasAccessedAfter(
                            processingInfoBuffer,
                            accessTimeThresholdBuf);

            boolean wasMatchFound;
            do {
                // Allow for task termination
                if (Thread.currentThread().isInterrupted()) {
                    // As we are outside of a txn the interruption is ok and everything will be in
                    // valid state for when purge is run again.
                    throw new TaskTerminatedException();
                }

                // With a read txn scan over all the proc info entries to find the next one that is ready for purge
                final Optional<RefStreamDefinition> optRefStreamDef = lmdbEnvironment.getWithReadTxn(readTxn ->
                        findNextRefStreamDef(
                                procInfoPooledBufferPair,
                                currRefStreamDefBufRef,
                                accessTimePredicate,
                                readTxn));

                if (optRefStreamDef.isPresent()) {
                    wasMatchFound = true;
                    final RefStreamDefinition refStreamDefinition = optRefStreamDef.get();

                    final RefStreamPurgeCounts refStreamPurgeCounts = purgeRefStreamIfEligible(
                            accessTimePredicate,
                            currRefStreamDefBufRef.get(),
                            refStreamDefinition);

                    // aggregate the counts
                    countsRef.getAndUpdate(counts ->
                            counts.increment(refStreamPurgeCounts));
                } else {
                    LOGGER.debug("No matching ref stream found");
                    wasMatchFound = false;
                }

                // Sleep block to slow things down for testing
//                try {
//                    Thread.sleep(20_000);
//                } catch (InterruptedException e) {
//                    Thread.currentThread().interrupt();
//                }
            } while (wasMatchFound);

            final PurgeCounts purgeCounts = countsRef.get();
            if (purgeCounts.refStreamDefsFailedCount == 0) {
                if (purgeCounts.isZero()) {
                    LOGGER.info("Nothing to purge in store '{}'", storeName);
                } else {
                    LOGGER.info(() -> LogUtil.message("Purge completed successfully on store '{}' - {}",
                            storeName, buildPurgeInfoString(startTime, purgeCounts)));
                }
            } else {
                // One or more ref stream defs failed
                throw new RuntimeException(LogUtil.message(
                        "Unable to purge {} ref stream definitions",
                        purgeCounts.refStreamDefsFailedCount));
            }
        } catch (TaskTerminatedException e) {
            // Expected behaviour so just rethrow, stopping it being picked up by the other
            // catch block
            LOGGER.debug("Purge terminated", e);
            LOGGER.warn(() -> "Purge terminated - " +
                              buildPurgeInfoString(startTime, countsRef.get()));
            throw e;
        } catch (Exception e) {
            LOGGER.error(() -> "Purge failed due to " + e.getMessage() + ". " +
                               buildPurgeInfoString(startTime, countsRef.get()), e);
            throw e;
        }

        //open a write txn
        //open a cursor on the process info table to scan all records
        //subtract purge age prop val from current time to give purge cut off ms
        //for each proc info record one test the last access time against the cut off time (without
        // de-serialising to long)
        //if it is older than cut off date then change its state to PURGE_IN_PROGRESS


        //process needs to be idempotent so we can continue a part finished purge. A new txn MUST always check the
        //processing info state to ensure it is still PURGE_IN_PROGRESS in case another txn has started a load, in which
        //case we won't purge. A purge txn must wrap at least the deletion of the key(range)/entry, the value
        // (if no other
        //refs). The deletion of the mapdef<=>uid paiur must be done in a txn to ensure consistency.
        //Each processing info entry should be be fetched with a read txn, then get a StripedSemaphore for the
        // streamdef
        //then open the write txn. This should stop any conflict with load jobs for that stream.

        //when overwrites happen we may have two values that had an association with same mapDef + key.  The
        // above process
        //will only remove the currently associated value.  We would have to scan the whole value table to look for


        // streamDef => mapDefs
        // mapDef => mapUID
        // mapUID => ValueKeys
        // ValueKey => value

        // <pipe uuid 2-18><pipe ver 2-18><stream id 8> => <create time 8><last access time 8><effective time 8><state1>
        // <pipe uuid 12-18><pipe ver 2-18><stream id 8><map name ?> => <mapUID 4>
        // <mapUID 4> => <pipe uuid 2-18><pipe ver 2-18><stream id 8><map name ?>
        // <mapUID 4><string Key ?> => <valueHash 4><id 2>
        // <mapUID 4><range start 8><range end 8> => <valueHash 4><id 2>
        // <valueHash 4><id 2> => <value type 1><reference count 4><value bytes ?>

        // increment ref count when
        // - putting new key(Range)/Value entry + new value entry (set initial ref count at 1)
        // - putting new key(Range)/Value entry with existing value entry
        // - overwrite key(Range)/Value entry (+1 on new value key)

        // decrement ref count when
        // - overwrite key(Range)/Value entry (-1 on old value key)
        // - delete key(Range)/Value entry

        // change to ref counter MUST be done in same txn as the thing that is making it change,
        // e.g the KV entry removal
    }
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