class ShardCleaner {

    @VisibleForTesting
    synchronized void cleanLocalShards()
    {
        // find all files on the local node
        Set<UUID> local = getLocalShards();

        // get shards assigned to the local node
        Set<UUID> assigned = dao.getNodeShards(currentNode, null).stream()
                .map(ShardMetadata::getShardUuid)
                .collect(toSet());

        // un-mark previously marked files that are now assigned
        for (UUID uuid : assigned) {
            shardsToClean.remove(uuid);
        }

        // mark all files that are not assigned
        for (UUID uuid : local) {
            if (!assigned.contains(uuid)) {
                shardsToClean.putIfAbsent(uuid, ticker.read());
            }
        }

        // delete files marked earlier than the clean interval
        long threshold = ticker.read() - localCleanTime.roundTo(NANOSECONDS);
        Set<UUID> deletions = shardsToClean.entrySet().stream()
                .filter(entry -> entry.getValue() < threshold)
                .map(Map.Entry::getKey)
                .collect(toSet());
        if (deletions.isEmpty()) {
            return;
        }

        for (UUID uuid : deletions) {
            deleteFile(storageService.getStorageFile(uuid));
            shardsToClean.remove(uuid);
        }

        localShardsCleaned.update(deletions.size());
        log.info("Cleaned %s local shards", deletions.size());
    }

    @Inject public  ShardCleaner(
            DaoSupplier<ShardDao> shardDaoSupplier,
            Ticker ticker,
            NodeManager nodeManager,
            StorageService storageService,
            Optional<BackupStore> backupStore,
            ShardCleanerConfig config);
    public  ShardCleaner(
            DaoSupplier<ShardDao> shardDaoSupplier,
            String currentNode,
            boolean coordinator,
            Ticker ticker,
            StorageService storageService,
            Optional<BackupStore> backupStore,
            Duration maxTransactionAge,
            Duration transactionCleanerInterval,
            Duration localCleanerInterval,
            Duration localCleanTime,
            Duration backupCleanerInterval,
            Duration backupCleanTime,
            int backupDeletionThreads,
            Duration maxCompletedTransactionAge);

}