Package org.exoplatform.addons.documents.cleanup.service

Class CleanupScanService

java.lang.Object
org.exoplatform.addons.documents.cleanup.service.CleanupScanService
@Service public class CleanupScanService extends Object
Asynchronous dry-run scan of the collaboration workspace, PARALLELISED over scan UNITS: the tree is partitioned once (see CleanupConstants.SPLIT_SCAN_ROOTS), several reader threads walk their own unit each, and ONE writer thread persists everything they stream through a bounded queue. Nothing is ever deleted here.

THE INVARIANT of the whole design is ONLY THE WRITER WRITES while readers are alive. It was first stated as "only the writer TOUCHES JPA", which was WRONG, and the mistake cost a Tribe-Dev dry run: a reader calls no cleanup storage bean but CleanupJcrStorage.scanRoot(java.lang.String, java.lang.String, int, org.exoplatform.addons.documents.cleanup.model.CleanupParams, org.exoplatform.addons.documents.cleanup.storage.CleanupJcrStorage.ScanBatchConsumer) — yet scanRoot resolves each CANDIDATE's owner through Social (getOwnerIdentityFromNodePath -> SpaceService / IdentityManager), and a cache MISS there runs a JPA query. On a reader thread with no container transaction that query failed with Session/EntityManager is closed; the per-node handler inside scanRoot swallowed it as a WARN, the unit still finished DONE, and every candidate whose space was not already cache-warm was silently dropped from the report. "No call to OUR storage" is not "no JPA": a Service of another domain is free to hit its own DAO, and this one does.

So each reader now opens its OWN read-only container transaction (readUnitTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, org.exoplatform.addons.documents.cleanup.model.CleanupScanUnit)), and the invariant that actually holds — the one worth defending, because it is what keeps write ordering and the checkpoint consistent — is about WRITES:

TERMINATION — what is guaranteed, and by WHICH mechanism. Nothing here bounds the DURATION of a scan: a dry-run over the target corpus spans hours to DAYS, is interrupted and resumed, and no mechanism below may ever cut a run short merely because it took long. What is bounded is BLOCKING WITHOUT PROGRESS, by four independent mechanisms, each covering what the previous one cannot:

  • a reader blocked on a full queue re-checks the stop flags between two bounded offers (QUEUE_TIMEOUT_MILLIS), so it stops as soon as the writer raised one — the fast path, and it covers only the failures that DO raise a flag
  • a writer thread that VANISHED raises no flag, so the coordinator does not INFER it from a timer: it observes Thread#isAlive() on the writer thread itself, at every slice of the reader wait (awaitReaders(java.util.concurrent.ExecutorService, org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, java.lang.Thread)). A dead writer stops the readers immediately, with nothing presumed
  • a writer that is ALIVE but permanently wedged (blocked on a database lock, say) is caught by an INACTIVITY watchdog: nothing drained for writerInactivityMillis WHILE batches wait in the queue. A silence bound and not a duration cap, and it can only fire while the writer has work it is not taking — a healthy scan, however long, never reaches it
  • the writer is told to stop by TWO independent signals, deliberately: the poison pill, delivered UNCONDITIONALLY (see postPoisonPill(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun)), and ScanRun#isStopped(), which the drain loop exits on even if no pill ever lands. Neither depends on the other
Every one of them leaves the campaign DRY_RUN_RUNNING and resumable from its unit checkpoints, lets the writer's frame unwind — so the container transaction it opened is closed — and releases the campaign id (the coordinator's finally) so the watchdog can relaunch the worker.

A FAILED UNIT IS RETRIED, BOUNDEDLY. A unit whose walk failed is not terminal: getUnitsToProcess hands it back to the next run, so the campaign is deliberately left DRY_RUN_RUNNING and the watchdog (CleanupCampaignService#resumeStalledWorkers) re-walks it — a transient JCR failure heals itself with no human in the loop. The bound is MAX_SCAN_UNIT_ATTEMPTS, and it bounds the WALK and not merely a counter: past it the subtree has proved unreadable, the unit is SETTLED-failed, CleanupScanUnitStorage#getUnitsToProcess stops handing it out — so no later run claims it, spends an attempt on it or re-walks it — and it stops holding the dry-run back. The report is then marked INCOMPLETE rather than silently partial (see completeCampaign(long, long)).

A campaign whose EVERY unit settled-failed (a JCR outage fails them all) is therefore not a spin either: its work list is empty, every watchdog tick is a cheap no-op that walks nothing, and the campaign stays visibly stuck in DRY_RUN_RUNNING — refused as a simulation, loudly logged, but never re-walked.

LEGACY CHECKPOINT — a campaign interrupted under the old SEQUENTIAL scheme carries a CHECKPOINT_PATH on its campaign row and has no unit row. On resume it is re-planned into units and that legacy checkpoint is IGNORED, so the scan re-walks the tree under the new partitioning. Deliberate and safe: saveCandidates de-duplicates on the campaign+nodeUuid unique constraint, so the replay costs work and can never duplicate a row.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final long
    MAX_SCAN_UNIT_ATTEMPTS
    FAILED walks a scan unit may spend IN TOTAL, the first walk included — THREE walks that each reached a FAILED outcome, not one walk plus three retries.

  • Constructor Summary

    Constructors
    Constructor
    Description
    CleanupScanService()
     

  • Method Summary

    Modifier and Type
    Method
    Description
    protected void
    drainQueue(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run)
    Writer body — the ONLY component writing what the scan produces.
    void
    drainQueueTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run)
    Transactional entry point of the WRITER thread, self-invoked from the runnable started by scanUnits(long, org.exoplatform.addons.documents.cleanup.model.CleanupParams, java.util.List<org.exoplatform.addons.documents.cleanup.model.CleanupScanUnit>, java.util.concurrent.ExecutorService, int, long, long): the project weaves @ContainerTransactional with AspectJ, which advises self-calls too — the same property executorService.execute(() -> scanTransactional(id)) already relies on.
    List<CleanupFailureGroup>
    getScanFailures(CleanupCampaign campaign)
    Grouped failures of a campaign's SCAN, one entry per distinct failure message code with the number of subtrees that carry it — the unit-level twin of CleanupCampaignService#getCampaignFailures, so the console renders both through the same block.
    CleanupScanUnitProgress
    getScanUnitProgress(CleanupCampaign campaign)
    Per-unit breakdown of a campaign's dry run.
    protected void
    postPoisonPill(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run)
    Hands the writer its terminal signal, UNCONDITIONALLY.
    protected void
    readUnit(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run, CleanupScanUnit unit)
    Reader body: walks ONE unit and posts everything it finds on the queue, WRITING nothing (it reads, through Social, to resolve owners — see readUnitTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, org.exoplatform.addons.documents.cleanup.model.CleanupScanUnit)).
    void
    readUnitTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run, CleanupScanUnit unit)
    Transactional entry point of a READER thread, self-invoked from the runnable submitted by scanUnits(long, org.exoplatform.addons.documents.cleanup.model.CleanupParams, java.util.List<org.exoplatform.addons.documents.cleanup.model.CleanupScanUnit>, java.util.concurrent.ExecutorService, int, long, long) — same AspectJ self-call property as drainQueueTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun).
    protected void
    restartCoordinatorTransaction()
    The COORDINATOR's own seam, third of the family for the same reason the reader's is separate from the writer's: it commits once per run — right before the terminal transition, so that write never flows through a persistence context as old as the run itself — and sharing the writer's seam would make every assertion on the writer's per-batch commit count silently include it.
    protected void
    restartReaderTransaction()
    The READER's twin of restartTransaction(), kept as a separate seam on purpose rather than reusing that one: the two commit on different threads for different reasons — the writer to make its batch durable, a reader only to let go of a pooled connection before it blocks (see commitThenPost(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanBatch)) — and one shared seam would make every assertion on the writer's commit count silently depend on how many batches the readers happened to emit.
    protected void
    restartTransaction()
    Commits the current batch and opens a fresh transaction.
    protected void
    scan(long campaignId)
    Scan coordinator, running as system (no conversation state needed), in four phases: PLAN the units, ESTIMATE their sizes in parallel, SCAN them with readers + one writer, then COMPLETE.
    void
    scanTransactional(long campaignId)
    Transactional entry point of the COORDINATOR: planning, per-unit estimation and the terminal transition run in ONE container transaction, exactly like the execution purge (CleanupExecutionService.executeCampaignTransactional(long)).
    void
    shutdown()
     
    void
    startScan(long campaignId)
    Starts (or resumes after a restart) the dry-run scan of a campaign.

    Methods inherited from class java.lang.Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Field Details

    • MAX_SCAN_UNIT_ATTEMPTS

      public static final long MAX_SCAN_UNIT_ATTEMPTS
      FAILED walks a scan unit may spend IN TOTAL, the first walk included — THREE walks that each reached a FAILED outcome, not one walk plus three retries. Bounds the whole unit retry: past the third recorded failure the subtree has proved unreadable whatever the cause was, and the watchdog must not re-walk it on every tick until the end of time.

      An attempt is spent when the WRITER RECORDS the FAILED outcome (CleanupScanUnitStorage#updateUnitFailure), never by the coordinator claiming the unit — the claim is a pure RUNNING flip. Spending it at claim time conflated interruptions with failures: a JVM restart (the EXPECTED event of a days-long scan) claimed every remaining unit again, including the hundreds the reader pool never reached, so two interruptions left every unit one real failure away from being settled with ZERO retries — the opposite of what this bound promises. A walk that never reaches an outcome (interrupt, abandoned reader, dead writer) now costs nothing, exactly like an interrupted purge item.

      See Also:

  • Constructor Details

    • CleanupScanService

      public CleanupScanService()

  • Method Details

    • shutdown

      @PreDestroy public void shutdown()

    • startScan

      public void startScan(long campaignId) throws org.exoplatform.commons.exception.ObjectNotFoundException
      Starts (or resumes after a restart) the dry-run scan of a campaign.
      Parameters:
      campaignId - campaign identifier
      Throws:
      org.exoplatform.commons.exception.ObjectNotFoundException - when the campaign doesn't exist

    • scanTransactional

      public void scanTransactional(long campaignId)
      Transactional entry point of the COORDINATOR: planning, per-unit estimation and the terminal transition run in ONE container transaction, exactly like the execution purge (CleanupExecutionService.executeCampaignTransactional(long)). The streamed candidate rows are NOT written here — they are written by the writer thread, which opens its own transaction (see drainQueueTransactional(ScanRun)) and commits per batch.

      It is a thin wrapper on purpose: the annotation is woven around THIS method, so the tests drive scan(long) directly and never boot a container to exercise the worker body — the annotation itself is pinned by reflection in CleanupScanServiceTest.

      Parameters:
      campaignId - campaign identifier

    • scan

      protected void scan(long campaignId)
      Scan coordinator, running as system (no conversation state needed), in four phases: PLAN the units, ESTIMATE their sizes in parallel, SCAN them with readers + one writer, then COMPLETE. Visible for tests, which invoke it directly instead of going through the transactional wrapper above.

    • readUnitTransactional

      public void readUnitTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run, CleanupScanUnit unit)
      Transactional entry point of a READER thread, self-invoked from the runnable submitted by scanUnits(long, org.exoplatform.addons.documents.cleanup.model.CleanupParams, java.util.List<org.exoplatform.addons.documents.cleanup.model.CleanupScanUnit>, java.util.concurrent.ExecutorService, int, long, long) — same AspectJ self-call property as drainQueueTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun).

      A reader needs a container transaction even though it writes NOTHING, because CleanupJcrStorage.scanRoot(java.lang.String, java.lang.String, int, org.exoplatform.addons.documents.cleanup.model.CleanupParams, org.exoplatform.addons.documents.cleanup.storage.CleanupJcrStorage.ScanBatchConsumer) resolves every candidate's owner through Social and a cache miss there is a JPA read (see the class comment for the failure this cost us). The transaction is therefore READ-ONLY, which is what makes restartReaderTransaction() safe to fail: a reader has no write to lose at commit.

      Parameters:
      run - state of the current scan run
      unit - the unit this reader walks

    • readUnit

      protected void readUnit(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run, CleanupScanUnit unit)
      Reader body: walks ONE unit and posts everything it finds on the queue, WRITING nothing (it reads, through Social, to resolve owners — see readUnitTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, org.exoplatform.addons.documents.cleanup.model.CleanupScanUnit)). The unit's own LAST_SCANNED_PATH is the resume position (null when it was never started).

      A JCR Session is NOT thread-safe, which is why per-thread sessions are MANDATORY here: scanRoot opens its own system session per call and holds it for its whole lazy walk, so every reader gets its own — never change that to share one.

      A failure marks THIS unit failed and nothing else: one unreadable subtree must not deny the whole simulation. The failure is posted, not written — only the writer writes.

    • drainQueueTransactional

      public void drainQueueTransactional(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run)
      Transactional entry point of the WRITER thread, self-invoked from the runnable started by scanUnits(long, org.exoplatform.addons.documents.cleanup.model.CleanupParams, java.util.List<org.exoplatform.addons.documents.cleanup.model.CleanupScanUnit>, java.util.concurrent.ExecutorService, int, long, long): the project weaves @ContainerTransactional with AspectJ, which advises self-calls too — the same property executorService.execute(() -> scanTransactional(id)) already relies on.
      Parameters:
      run - state of the current scan run

    • drainQueue

      protected void drainQueue(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run)
      Writer body — the ONLY component writing what the scan produces. Drains the queue until the poison pill, and per batch: persists the candidates, checkpoints THE UNIT THE BATCH CAME FROM, recomputes the aggregate progress and its ETA, pushes the progress event, then commits.

      It exits on FOUR signals, and the first two are deliberately redundant: the poison pill, isStopped() — the run's own stop flags, so the writer stops even if no pill ever lands on a full queue — the abort re-read, and a failure. The pill alone was not enough: the coordinator raised writerFailed to unblock the readers, post then refused to enqueue the pill precisely BECAUSE the run was stopped, and this loop, which consulted the pill and the abort but never the stop flags, saw a campaign still legitimately DRY_RUN_RUNNING and looped forever — leaking the thread and the container transaction its entry point had opened.

      Every drained envelope also stamps ScanRun#markDrained(), the marker the coordinator's inactivity watchdog reads (awaitReaders(java.util.concurrent.ExecutorService, org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, java.lang.Thread)): draining IS the progress signal of this design, and a writer that stops stamping it while batches wait for it is a wedged writer.

      The aggregate numerator is kept IN MEMORY (seeded from the persisted per-unit counts) rather than re-summed per batch: the sum is already known exactly, and a database aggregate per batch would be pure overhead on a multi-million-node walk.

      Visible for tests, which drive it directly so the woven aspect above never boots a container.

    • getScanFailures

      public List<CleanupFailureGroup> getScanFailures(CleanupCampaign campaign)
      Grouped failures of a campaign's SCAN, one entry per distinct failure message code with the number of subtrees that carry it — the unit-level twin of CleanupCampaignService#getCampaignFailures, so the console renders both through the same block.

      Gated on the campaign's own INCOMPLETE verdict, and NOT on the live unit rows: a unit that failed while attempts remain is a transient failure the watchdog is already re-walking, and surfacing it would report a subtree as missing from a report that is not even finished. Only a scan the coordinator RECORDED as incomplete has anything to show here.

      Every group is retryable = false, which is the honest answer: these subtrees exhausted MAX_SCAN_UNIT_ATTEMPTS walks, and there is deliberately no console retry for them — re-running the scan of a campaign that already left DRY_RUN_RUNNING is not an edge this lifecycle has.

      Parameters:
      campaign - campaign whose scan failures are wanted
      Returns:
      the groups, EMPTY when the campaign's scan covered the whole tree

    • getScanUnitProgress

      public CleanupScanUnitProgress getScanUnitProgress(CleanupCampaign campaign)
      Per-unit breakdown of a campaign's dry run.

      Deliberately NOT gated on anything, unlike getScanFailures(org.exoplatform.addons.documents.cleanup.model.CleanupCampaign): that one reports subtrees definitively missing from a finished report, so it must wait for the verdict; this one exists to be read WHILE the scan runs, and gating it would remove it exactly when it is needed. A scan whose node percentage sits at 100% with the campaign still DRY_RUN_RUNNING is either re-walking a unit or held open by one, and only these counts tell an administrator which.

      Parameters:
      campaign - campaign whose unit breakdown is wanted
      Returns:
      the breakdown, zeroed for a campaign whose units were never planned

    • postPoisonPill

      protected void postPoisonPill(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun run)
      Hands the writer its terminal signal, UNCONDITIONALLY. It must not go through post(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanBatch), which gives up as soon as isStopped() answers true: the flag raised to unblock the readers was then the very flag that kept the writer from ever being told to stop — the pill was never enqueued, and the writer spun on an empty queue forever, holding the container transaction its @ContainerTransactional entry point had opened. One leaked transaction per killed scan.

      The queue MAY BE FULL here, so the delivery must not block forever either. Two cases, and they are not symmetric:

      • the run is STOPPED: the queued batches are abandoned whatever happens — nothing will checkpoint them, and their nodes are simply re-walked on the next resume — so the queue is CLEARED to make room. Dropping work that is already forfeit is the cheap half of the trade
      • the run is NOT stopped: those batches are real work the writer is still expected to persist, so they are never dropped. Every reader has finished by now, so the queue can only SHRINK — one drained envelope is all the pill needs — and the offers are therefore retried for as long as the writer keeps DRAINING, on the very definition of wedged the watchdog uses (isWriterWedged(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun)) rather than on a deadline of its own. A healthy but slow writer is never given up on; a wedged one is, and the drain loop's isStopped() exit stops it instead (see drainQueue(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun))
      Parameters:
      run - state of the current scan run

    • restartTransaction

      protected void restartTransaction()
      Commits the current batch and opens a fresh transaction. Extracted as a single call so the worker tests — which drive the writer body directly, with no PortalContainer booted — can neutralize it, exactly like CleanupExecutionService.restartTransaction().

    • restartReaderTransaction

      protected void restartReaderTransaction()
      The READER's twin of restartTransaction(), kept as a separate seam on purpose rather than reusing that one: the two commit on different threads for different reasons — the writer to make its batch durable, a reader only to let go of a pooled connection before it blocks (see commitThenPost(org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanRun, org.exoplatform.addons.documents.cleanup.service.CleanupScanService.ScanBatch)) — and one shared seam would make every assertion on the writer's commit count silently depend on how many batches the readers happened to emit.

    • restartCoordinatorTransaction

      protected void restartCoordinatorTransaction()
      The COORDINATOR's own seam, third of the family for the same reason the reader's is separate from the writer's: it commits once per run — right before the terminal transition, so that write never flows through a persistence context as old as the run itself — and sharing the writer's seam would make every assertion on the writer's per-batch commit count silently include it.