feat(reconcile): walk Proton once the last walk is a week old

CLAUDE.md

@@ -8,9 +8,10 @@ this mirror includes. This file is what a change must not break.
 
 Nothing in this repo starts a run: an external scheduler dispatches `sync.yml` nightly.
 This mirror includes the toolbox alone: `Taskfile.yml` owns the order, one
-`python -m migrator <command>` per step, and `src/migrator/` owns every decision. The
-infrastructure modules there are from donphi/dropbox_proton at `cfd0e57`, MIT; the phases
-under `src/migrator/phases/` and the Taskfile are this repository's own.
+`python -m migrator <command>` per step, and `src/migrator/` owns every decision but when
+to walk Proton, which is lib's `due`. The infrastructure modules there are from
+donphi/dropbox_proton at `cfd0e57`, MIT; the phases under `src/migrator/phases/` and the
+Taskfile are this repository's own.
 
 ## Must knows
 
@@ -42,8 +43,10 @@ under `src/migrator/phases/` and the Taskfile are this repository's own.
 - **The logs are public.** The report is built from the state and carries counts only;
   errors print as their class unless `MIRROR_VERBOSE=1`; `op run` masks every value.
 - **Run flags go after the double dash** (`task sync -- RUN_BUDGET_MIN=30 RECONCILE=true`,
-  or the workflow's `vars` input); the Taskfile maps them to the environment the migrator
-  reads. `RECONCILE` takes the literal word `true`.
+  or the workflow's `vars` input). The Taskfile maps `RUN_BUDGET_MIN` to the environment
+  the migrator reads; `RECONCILE` (`true`, `false` or `auto`) is lib's `due`'s, which
+  leaves `.run/reconcile` for the migrator when the last complete walk is
+  `RECONCILE_HOURS` (168) old.
 - **`config/mirror.toml` is strict** and rejects unknown keys; the three account
   identifiers come from the environment and override its keys.
 

README.md

@@ -76,7 +76,7 @@ flowchart LR
 | `confirm` | The upload summary must account for every verified file plus every folder, and every failure must name a file in the batch. Those alone are recorded as failed; the rest confirm |
 | `checkpoint` | Merges the confirmed rows into `mirror_objects` and pushes the state to the bucket, a dated copy first and then the canonical key. Always the last step of a batch, so a killed run repeats at most one |
 | `trash` | Only when every planned batch landed. A topmost folder the mirror holds nothing live under goes in one `filesystem trash` call, subtree and all; a folder still holding live files gets its deleted files trashed by name, 50 paths per call. Each unit's `mirror_objects` rows are dropped as it lands. Checkpoints every 50 units, stops at the run budget and chains the next run for the rest |
-| `reconcile` | On the first run of the configured weekday, or with `RECONCILE=true`: a full Proton walk compared against `mirror_objects`. Rows Proton lacks or mis-sizes are dropped so they re-upload; nodes neither Dropbox nor the state knows are trashed. A walk that does not fit one run resumes on the next, and a partial walk drops and trashes nothing |
+| `reconcile` | Once the last complete walk is `RECONCILE_HOURS` (168, a week) old, by lib's `due`, or with `RECONCILE=true`: a full Proton walk compared against `mirror_objects`. Rows Proton lacks or mis-sizes are dropped so they re-upload; nodes neither Dropbox nor the state knows are trashed. A walk that does not fit one run resumes on the next, and a partial walk drops and trashes nothing |
 | `report` | Builds the step summary from the state alone, finishes the run row, writes the chain marker, and returns the run's status |
 
 Every step is plan-by-default: `batches`, `trash`, `reconcile`, `report` and
@@ -107,6 +107,7 @@ The bucket holds the state and the session, nothing of the mirrored tree.
 .state/state.sqlite.xz.age                     the state: evidence tables, mirror_objects, runs, batches, deletions
 .state/history/<epoch>-<label>.sqlite.xz.age   one copy per checkpoint; label is the batch number, trash or trash-<folders>, reconcile or report
 .state/session.tar.age                         the Proton CLI session; no history, a stale copy cannot be restored
+.state/reconciled                              the start epoch of the run that last completed a Proton walk, plain text
 ```
 
 | What | Why |
@@ -196,7 +197,7 @@ task status                        # counts and the last run's figures from the
 task test && task lint             # pytest; ruff check and format check
 task sync                          # one budgeted run, the same thing Actions runs
 task sync -- RUN_BUDGET_MIN=30     # a shorter budget
-task sync -- RECONCILE=true        # force the weekly Proton walk (the literal word true)
+task sync -- RECONCILE=true        # force the weekly Proton walk (true, false or auto)
 task state-rollback                # list the dated history objects
 task state-rollback -- <key>       # copy one of them over the canonical state
 task session-seal -- .run/pd       # encrypt a laptop Proton CLI session into the bucket
@@ -234,10 +235,10 @@ older than seven days are pruned, the same clock as the bucket's history copies.
 - **A run stops on budget every night.** Lower `batch_files` or `batch_gb`; the throughput
   rows say which. A run that checkpointed nothing does not chain and fails instead.
 - **Proton 429s.** The throttling table is the gauge; lower `batch_gb`.
-- **A weekly reconcile does not finish in one run.** Normal on a large tree; it resumes on
-  the next run that reconciles.
+- **A weekly reconcile does not finish in one run.** Normal on a large tree; only a
+  complete walk is recorded, so the next run is due again and resumes it.
 - **A flag seems ignored.** Flags go after the double dash; before it they set a host-side
-  var that never reaches the container. `RECONCILE` takes the literal word `true`.
+  var that never reaches the container. `RECONCILE` takes `true`, `false` or `auto`.
 - **A case-only rename in Dropbox does not reach Proton.** Files are keyed by lowercased
   path. Rename to something else and back if the case matters.
 - **Move the folder in Proton.** Move it anywhere under My files and change the vault's
@@ -264,7 +265,6 @@ unknown keys. It names no account.
 | `budget.disk_headroom_gb` | Free disk the runner must keep beyond a batch's staging |
 | `budget.listing_floor_ratio` | Refuse a listing smaller than this share of the mirrored file count |
 | `proton.walk_workers` | Folder listings in flight during the reconcile walk, each from its own copy of the session |
-| `reconcile.weekday` | UTC weekday (0 is Monday) whose first run does the Proton walk |
 
 The three account identifiers in `op.env` override the TOML keys
 `dropbox.expected_account_id`, `proton.destination` and `proton.expected_destination_uid`,

Taskfile.yml

@@ -14,16 +14,17 @@ version: '3'
 vars:
   RUN_EPOCH:
     sh: date +%s
+  RECONCILE_HOURS: 168   # the full Proton walk, weekly; lib's toolbox `due` counts from the last complete one
 
-# Run-scoped environment for every migrator command. RECONCILE and RUN_BUDGET_MIN are
-# task vars a run sets after the double dash (task sync -- RECONCILE=true
-# RUN_BUDGET_MIN=30, or the workflow's vars input), mapped to the names the migrator
-# reads; unset, they render empty, which it reads as unset. The migrator's work
-# directory and config path are its own defaults, .run and config/mirror.toml, and
-# AWS_REGION is an ENV line in the image, so every in-toolbox command sees it.
+# Run-scoped environment for every migrator command. RUN_BUDGET_MIN is a task var a run
+# sets after the double dash (task sync -- RUN_BUDGET_MIN=30, or the workflow's vars
+# input), mapped to the name the migrator reads; unset, it renders empty, which it reads
+# as unset. RECONCILE, set the same way, is read by the toolbox's `due`, not the
+# migrator. The migrator's work directory and config path are its own defaults, .run and
+# config/mirror.toml, and AWS_REGION is an ENV line in the image, so every in-toolbox
+# command sees it.
 env:
   MIRROR_RUN_EPOCH: '{{.RUN_EPOCH}}'
-  RECONCILE: '{{.RECONCILE}}'
   RUN_BUDGET_MIN: '{{.RUN_BUDGET_MIN}}'
   PROTON_DRIVE_LOG_LEVEL: INFO   # the Proton CLI's own log level; its lines reach the public job log
 
@@ -61,6 +62,7 @@ tasks:
     desc: The full budgeted run, inside the toolbox
     cmds:
       - task: clock
+      - task: due
       - task: clock-phase
       - task: session
       - task: state
@@ -78,6 +80,7 @@ tasks:
     desc: Read-only half of the pipeline, inside the toolbox; prints the report it would make
     cmds:
       - task: clock
+      - task: due
       - task: clock-phase
       - task: session
       - task: state
@@ -97,7 +100,29 @@ tasks:
   plan-phase: { internal: true, cmds: ['python -m migrator plan'] }
   batches:    { internal: true, cmds: ['python -m migrator --apply batches'] }
   trash:      { internal: true, cmds: ['python -m migrator --apply trash'] }
-  reconcile:  { internal: true, cmds: ['python -m migrator --apply reconcile'] }
+  reconcile:
+    internal: true
+    cmds:
+      - python -m migrator --apply reconcile
+      - task: record-walk
+  # Only a complete walk is recorded; a partial one leaves the next run due, and it resumes.
+  record-walk:
+    internal: true
+    status: ['test ! -f {{.RUN}}/walked']
+    cmds: [{ task: reconciled }]
+  # What lib's toolbox `due` and `reconciled` keep .state/reconciled through, as an engine's
+  # pull and push would: one plain object of the bucket, over the image's s3.
+  pull:
+    internal: true
+    requires: { vars: [KEY, OUT, MISSING] }
+    cmds:
+      - |
+        rc=0; s3 get {{.KEY}} {{.OUT}} || rc=$?
+        case $rc in 0) ;; 3) {{.MISSING}} ;; *) exit $rc ;; esac
+  push:
+    internal: true
+    requires: { vars: [FILE, KEY] }
+    cmds: ['s3 put {{.FILE}} {{.KEY}}']
   # Builds .run/report.md from the state, closes the run row, writes the chain marker.
   report-phase: { internal: true, cmds: ['python -m migrator --apply report'] }
   empty-trash-pipeline:

config/mirror.toml

@@ -35,6 +35,3 @@ listing_floor_ratio = 0.5
 # The reconcile walk is one `filesystem list` process per folder; this many run at once,
 # each from its own copy of the CLI session. Eight matches dropbox.download_workers.
 walk_workers = 8
-
-[reconcile]
-weekday = 0

render.txt

@@ -1,4 +1,21 @@
-task: [clock] mkdir -p /work/.run && rm -f /work/.run/chain && date -u '+%s %H %u' > /work/.run/start.txt
+task: [clock] mkdir -p /work/.run && rm -f /work/.run/chain /work/.run/reconcile && date -u +%s > /work/.run/start.txt
+task: [pull] rc=0; s3 get .state/reconciled /work/.run/reconciled || rc=$?
+case $rc in 0) ;; 3) : > /work/.run/reconciled ;; *) exit $rc ;; esac
+
+task: [due-rule] rm -f /work/.run/reconcile
+r=auto; h=168
+now=$(cat /work/.run/start.txt); last=$(cat /work/.run/reconciled 2>/dev/null || true)
+case $last in ''|0?*|*[!0-9]*) last=0 ;; esac
+case $h in ''|0*|*[!0-9]*) echo "due: RECONCILE_HOURS is a whole number of hours from 1, not $h" >&2; exit 1 ;; esac
+case $r in
+  true) touch /work/.run/reconcile ;;
+  false) ;;
+  auto) test $(( now - last )) -lt $(( h * 3600 - 1800 )) || touch /work/.run/reconcile ;;
+  *) echo "due: RECONCILE is true, false or auto" >&2; exit 1 ;;
+esac
+test "$last" = 0 && ago="none on record" || ago="$(( (now - last) / 3600 )) h ago"
+echo "due: RECONCILE=$r RECONCILE_HOURS=$h, last reconcile $ago: $(test -f /work/.run/reconcile && echo reconciling || echo not this run)"
+
 task: [clock-phase] python -m migrator clock
 task: [session] python -m migrator session
 task: [state] python -m migrator state
@@ -8,6 +25,8 @@ task: [plan-phase] python -m migrator plan
 task: [batches] python -m migrator --apply batches
 task: [trash] python -m migrator --apply trash
 task: [reconcile] python -m migrator --apply reconcile
+task: [reconciled] cp /work/.run/start.txt /work/.run/reconciled
+task: [push] s3 put /work/.run/reconciled .state/reconciled
 task: [report-phase] python -m migrator --apply report
 task: [report-mirror] test -f /work/.run/report.md || python -m migrator --apply report || true
 task: [report-mirror] test ! -f /work/.run/report.md || cat /work/.run/report.md >> "${GITHUB_STEP_SUMMARY:-/dev/stdout}"

src/migrator/commands.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 
 from . import session, statefile
-from .config import Config, load_config
+from .config import load_config
 from .env import Runtime
 from .paths import WorkPaths
 from .state import State
@@ -25,7 +25,7 @@ def clock(runtime: Runtime, args: list[str]) -> int:
     paths = _paths(runtime)
     shutil.rmtree(paths.staging, ignore_errors=True)
     paths.staging.mkdir()
-    for stale in (paths.report, paths.chain):
+    for stale in (paths.report, paths.chain, paths.walked):
         stale.unlink(missing_ok=True)
     started = datetime.fromtimestamp(runtime.run_epoch, UTC)
     stamp = {
@@ -42,23 +42,6 @@ def read_clock(paths: WorkPaths) -> dict[str, int]:
     return json.loads(paths.clock.read_text(encoding="utf-8"))
 
 
-def is_reconcile_run(
-    cfg: Config, runtime: Runtime, db: State, *, start_epoch: int, weekday: int
-) -> bool:
-    """RECONCILE=true, or the first run that starts on the configured UTC weekday. Keyed
-    on the day, not an hour: chained and queued runs start at any hour."""
-    if runtime.reconcile:
-        return True
-    if weekday != cfg.reconcile.weekday:
-        return False
-    day_start = start_epoch - start_epoch % 86400
-    earlier = db.connection.execute(
-        "SELECT COUNT(*) FROM runs WHERE start_epoch >= ? AND start_epoch < ?",
-        (day_start, start_epoch),
-    ).fetchone()[0]
-    return int(earlier) == 0
-
-
 def session_restore(runtime: Runtime, args: list[str]) -> int:
     paths = _paths(runtime)
     session.restore(runtime, paths, Store(runtime, paths))
@@ -81,17 +64,15 @@ def state(runtime: Runtime, args: list[str]) -> int:
     db = State(paths.state_db, cfg.mirror.id)
     try:
         db.initialize_migration(cfg.source_file, cfg.source_sha256)
-        # Decided before this run's row exists, so the row itself cannot count as "earlier".
-        reconcile = is_reconcile_run(
-            cfg, runtime, db, start_epoch=stamp["start_epoch"], weekday=stamp["weekday"]
-        )
         run_id = db.start_run(
             start_epoch=stamp["start_epoch"],
             hour_utc=stamp["hour_utc"],
             weekday=stamp["weekday"],
             budget_minutes=runtime.budget_override or cfg.budget.run_budget_minutes,
             host=runtime.host,
-            reconcile=reconcile,
+            # lib's toolbox `due` decided before this command, by the age of the last
+            # complete walk.
+            reconcile=paths.reconcile.exists(),
         )
         files, size = db.mirror_totals()
     finally:

src/migrator/config.py

@@ -168,24 +168,18 @@ def headroom_bytes(self) -> int:
         return round(self.disk_headroom_gb * 1024**3)
 
 
-@dataclass(frozen=True)
-class Reconcile:
-    weekday: int = 0  # the first run that starts on this UTC weekday walks Proton
-
-
 @dataclass(frozen=True)
 class Config:
     mirror: Mirror
     dropbox: Dropbox
     proton: Proton
     budget: Budget
-    reconcile: Reconcile
     source_file: Path
     source_sha256: str
 
 
 _MIRROR_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
-SECTIONS = {"mirror", "dropbox", "proton", "budget", "reconcile"}
+SECTIONS = {"mirror", "dropbox", "proton", "budget"}
 # Account-specific values live in the vault (op.env) and override the file, so
 # the committed configuration names no account.
 ENV_OVERRIDES = {
@@ -216,7 +210,6 @@ def load_config(path: str | Path, environ: Mapping[str, str] | None = None) -> C
         dropbox=_section(Dropbox, data.get("dropbox", {}), base, "dropbox"),
         proton=_section(Proton, data.get("proton", {}), base, "proton"),
         budget=_section(Budget, data.get("budget", {}), base, "budget"),
-        reconcile=_section(Reconcile, data.get("reconcile", {}), base, "reconcile"),
         source_file=source,
         source_sha256=sha256_file(source),
     )
@@ -276,5 +269,3 @@ def validate_config(cfg: Config) -> None:
     _nonnegative(cfg.budget.disk_headroom_gb, "budget.disk_headroom_gb")
     if not 0 < cfg.budget.listing_floor_ratio <= 1:
         raise ConfigError("budget.listing_floor_ratio must be in (0, 1]")
-    if type(cfg.reconcile.weekday) is not int or not 0 <= cfg.reconcile.weekday <= 6:
-        raise ConfigError("reconcile.weekday must be 0 (Monday) to 6 (Sunday)")

src/migrator/env.py

@@ -23,7 +23,6 @@ class Runtime:
     config_path: Path
     run_epoch: int | None
     budget_override: int | None
-    reconcile: bool
     verbose: bool
     r2_bucket: str
     age_identity: str
@@ -46,7 +45,6 @@ def from_environ(cls, environ: Mapping[str, str] | None = None) -> Runtime:
             config_path=Path(env.get("MIRROR_CONFIG", "config/mirror.toml")),
             run_epoch=_int(env, "MIRROR_RUN_EPOCH"),
             budget_override=_int(env, "RUN_BUDGET_MIN"),
-            reconcile=env.get("RECONCILE", "").lower() == "true",
             verbose=env.get("MIRROR_VERBOSE", "") == "1",
             r2_bucket=env.get("MIRROR_R2_BUCKET", ""),
             age_identity=env.get("MIRROR_AGE_IDENTITY", ""),

src/migrator/paths.py

@@ -42,6 +42,16 @@ def report(self) -> Path:
     def chain(self) -> Path:
         return self.root / "chain"
 
+    @property
+    def reconcile(self) -> Path:
+        # lib's toolbox `due` leaves it when this run is to walk Proton.
+        return self.root / "reconcile"
+
+    @property
+    def walked(self) -> Path:
+        # A complete walk leaves it, and the Taskfile then records the reconcile in R2.
+        return self.root / "walked"
+
     @property
     def age_key(self) -> Path:
         return self.root / "age.key"

src/migrator/phases/p60_reconcile.py

@@ -220,6 +220,7 @@ def run(ctx: PhaseContext) -> PhaseResult:
     if stray_folders:
         proton.trash(stray_folders, PHASE)
     statefile.push(ctx.state, ctx.runtime, ctx.paths, store, label=label)
+    ctx.paths.walked.touch()
     ctx.logger.info(
         PHASE,
         "figures",

tests/conftest.py

@@ -7,7 +7,7 @@
 import pytest
 
 from migrator import crypt, session
-from migrator.config import Budget, Config, Dropbox, Mirror, Proton, Reconcile
+from migrator.config import Budget, Config, Dropbox, Mirror, Proton
 from migrator.env import Runtime
 from migrator.logging import RunLogger
 from migrator.paths import WorkPaths
@@ -66,7 +66,6 @@ def make(tmp_path: Path, **overrides) -> Config:
                 ),
             ),
             budget=overrides.get("budget", Budget()),
-            reconcile=overrides.get("reconcile", Reconcile()),
             source_file=source,
             source_sha256="test-config-hash",
         )