feat(kudos): legacy sub-ranges become eligible on their own age

- select_next_unit catalogues the legacy partition as soon as it is attached and takes a unit when its own upper bound is past the window, so the catch-up of pre-swap history starts when retention is enabled instead of one window after the swap time. The partition drop still waits for every sub-range, so the last one, ending at the swap time, holds the detach for a window.
- retention_health computes oldest_eligible_age by sub-range for a legacy partition with catalog rows (_eligible_uppers), and by partition bound otherwise.
- Tests: the early catalogue and the per-sub-range selection.
- Docs: the how-to, reference, explanation and ADR 42 state the per-unit rule.

docs/decisions/0042-roll-up-statistics-and-archive-currency-after-the-online-window.md

@@ -81,9 +81,9 @@ therefore does not need a writer barrier, sequence sealing or a drain under the
 The primary key becomes (`id`, `created`). No foreign key references either table. The existing heap becomes the
 first partition. A validated `CHECK` on `created` lets it attach without a scan, the swap is a rename under a brief
 exclusive lock with no copy, and its indexes attach to partitioned indexes created on the parent, including the partial
-unapplied and quarantine indexes. New partitions default to one week. The first partition becomes eligible 30 days
-after the swap. Because it holds the whole pre-swap history, its replacements are built in bounded `created`
-sub-ranges recorded in the catalog, and the single drop waits for all of them. The migration is rehearsed on a restored
+unapplied and quarantine indexes. New partitions default to one week. Because the first partition holds the whole
+pre-swap history, its replacements are built in bounded `created` sub-ranges recorded in the catalog, each eligible
+on its own age, and the single drop waits for all of them, so it happens one window after the swap. The migration is rehearsed on a restored
 copy before production.
 
 The retention job runs under its own advisory lock and processes one eligible partition at a time in ascending order.

docs/explanation/kudos_retention.md

@@ -59,8 +59,9 @@ the decision and the other options it weighed.
 ## Partitions by age
 
 Both event tables are partitioned by `RANGE (created)`. `created` is set when the row is written, so a row that
-commits late still lands in the partition of its write time. A partition is eligible for removal once its upper bound
-is older than its dataset's window. A transaction that is still open cannot write into a partition whose upper bound
+commits late still lands in the partition of its write time. A unit is eligible for removal once its upper bound
+is older than its dataset's window: a weekly partition as a whole, and the legacy partition sub-range by sub-range,
+with its drop waiting for the last one. A transaction that is still open cannot write into a range whose upper bound
 is a whole window in the past, so eligibility by age does not need a writer barrier, sequence sealing or a drain under
 the mode gate. The drop still recounts the rows, because a clock fault or a reference time moved forward on a live
 database can put a row where the age argument says none can be.

docs/how-to/kudos_ledger_operations.md

@@ -317,14 +317,17 @@ docker exec -w /app -e PYTHONPATH=/app <container> python tools/kudos_ledger_adm
    `"detail": "no actionable unit"`, or a step result. It prints `"outcome": "skipped_lock"` when the scheduled tick
    holds the lock, and then it can be run again. While retention is disabled, the detail is `retention disabled`. The
    command runs a real tick and can drop a partition.
-3. Expect the first ticks to be idle. A partition becomes eligible when its upper bound is older than the window, so
-   the legacy partition becomes eligible one window after `swap_ts`. Until then `oldest_eligible_age_seconds` is null
-   for both datasets and the gauge reads 0.
-4. Watch the legacy partition being processed. The first eligible tick catalogues it as one unit per partition width,
-   starting at the Monday on or before its oldest row, and logs
-   `Kudos retention catalogued <n> units of <dataset> partition <table>_p_legacy`. Each later tick makes one transition
+3. Expect the first tick to catalogue the legacy partition. A unit becomes eligible when its upper bound is older
+   than the window. A weekly partition is one unit, catalogued when its upper bound is older than the window. The
+   legacy partition is catalogued as soon as the tick sees it, as one unit per partition width starting at the Monday
+   on or before its oldest row, and each of its sub-ranges becomes eligible on its own age. The tick logs
+   `Kudos retention catalogued <n> units of <dataset> partition <table>_p_legacy`. `oldest_eligible_age_seconds` is
+   then the age of the oldest sub-range past the window, or null when the legacy partition holds less than a window
+   of history.
+4. Watch the legacy partition being processed. Each tick makes one transition
    across both datasets, and the scheduler runs a tick every 60 seconds. Statistics units move through `rolled_up`
-   and `dumped`, and currency units through `checkpointed`, `exported` and `verified`. The legacy partition is
+   and `dumped`, and currency units through `checkpointed`, `exported` and `verified`. The last sub-range, which
+   ends at `swap_ts`, becomes eligible one window after `swap_ts`. The legacy partition is
    dropped by one tick after every one of its units is ready. Until then it stays attached, so
    `oldest_eligible_age_seconds` keeps growing and `pending_units` stays above zero. Verify: `retention status` shows
    the units advancing, `last_step_age_seconds` stays near the tick interval, and `horde.kudos.retention.steps` grows
@@ -402,8 +405,10 @@ currency unit recomputed over a recorded boundary blocks with `boundary already
 
 ## Rehearsing partition removal on a restored copy
 
-Nothing becomes eligible until a partition's upper bound is older than the window, so a freshly converted database
-has nothing to remove for at least the window. To exercise the removal path before that, restore a copy of the
+A legacy sub-range becomes eligible when its upper bound is older than the window, and a weekly partition when its
+own is, so a freshly converted database with less than a window of history does not remove anything for at least
+the window, and one with more history does not drop anything before the window either, because the legacy drop waits
+for its last sub-range. To exercise the removal path before that, restore a copy of the
 database, point it at a scratch bucket or at a prefix of its own (`KUDOS_ARCHIVE_PREFIX`), enable retention, and move
 the clock forward for one tick at a time:
 

docs/reference/kudos_accounting.md

@@ -439,9 +439,11 @@ A unit's `source_range` is `<partition_name>/<lower>_<upper>`. Each bound is for
 the unit they came from. A retried step deletes the rows with its own `source_range` before inserting them again. A UTC
 day can span two units, so readers sum across `source_range`.
 
-A partition is eligible when its upper bound is older than the dataset's window. A tick creates the catalog rows of an
-eligible partition, in state `pending`, the first time it selects that partition. Units are processed in order of
-upper bound, oldest first.
+A unit is eligible when its upper bound is older than the dataset's window. A tick creates the catalog rows of a
+weekly partition, in state `pending`, once its upper bound is older than the window, and the rows of the legacy
+partition the first time it sees it attached, because the legacy sub-ranges age one by one and the partition's own
+upper bound, the swap time, is the bound of its last sub-range only. Units are processed in order of upper bound,
+oldest first.
 
 ### Retention states
 
@@ -707,7 +709,7 @@ The tier sizes from the same listing are the `horde.kudos.archive.tier_bytes` va
 | --- | --- | --- | --- |
 | `horde.kudos.retention.steps` | Counter, `1` | `horde.kudos.dataset`, `horde.kudos.retention.step`, `horde.kudos.retention.outcome` | One per tick. The step is the `RetentionStep` that ran, or `none`. The outcome is a `RetentionTickOutcome`: `skipped_lock`, `idle`, `stepped`, `blocked`, `failed`, `refused` or `deferred`. Each tick adds 0 to the `blocked` and `failed` series of both datasets, so those series exist before the first event |
 | `horde.kudos.retention.dump_failures` | Counter, `{failure}` | `horde.kudos.dataset` | One per failed statistics dump attempt, including the one that blocks the unit. Each tick adds 0 for both datasets |
-| `horde.kudos.retention.oldest_eligible_age` | Gauge, `s` | `horde.kudos.dataset` | How long the oldest partition still attached has been past its window. 0 when every attached partition is inside its window |
+| `horde.kudos.retention.oldest_eligible_age` | Gauge, `s` | `horde.kudos.dataset` | How long the oldest unit still online has been past its window, by sub-range for the legacy partition. 0 when every unit is inside its window |
 | `horde.kudos.retention.blocked_units` | Gauge, `{unit}` | `horde.kudos.block_reason` | Catalog units in `blocked`, one series per `RetentionBlockReason` value, 0 when no unit has that reason |
 | `horde.kudos.retention.partition_coverage` | Gauge, `1` | None | 1 when each event table has a partition containing the real current time and one containing it plus one partition width, else 0 |
 | `horde.kudos.retention.held_partitions` | Gauge, `{partition}` | None | `held_partitions` from `retention_health`. 0 when `KUDOS_RETENTION_HOLD_DROPS` is not `true` |
@@ -733,7 +735,7 @@ does not count it. Other retention log lines are not under contract.
 
 | Field | Meaning |
 | --- | --- |
-| `oldest_eligible_age_seconds` | Per dataset, how long the oldest partition still attached has been past its window, or null |
+| `oldest_eligible_age_seconds` | Per dataset, how long the oldest unit still online has been past its window, by sub-range for the legacy partition, or null |
 | `blocked_units` | Catalog units in `blocked` |
 | `blocked_units_by_reason` | Catalog units in `blocked` per `RetentionBlockReason` value, with every reason present |
 | `pending_units` | Catalog units neither `dropped` nor `blocked`, including currency units waiting for an archive and `detached` units |

horde/database/kudos_retention.py

@@ -730,10 +730,13 @@ def select_next_unit(
     hold_drops: bool = False,
     detached_drop_due_before: datetime | None = None,
 ) -> NextUnit:
-    """Return the oldest actionable unit of one dataset, or the oldest eligible partition that has no catalog rows.
+    """Return the oldest actionable unit of one dataset, or the oldest partition that needs catalog rows.
 
-    A partition is eligible when its upper bound is before ``cutoff``. Units are taken in upper-bound order and skipped
-    while they cannot move:
+    A unit is eligible when its upper bound is before ``cutoff``. A weekly partition has one unit with the partition's
+    bounds, so it is catalogued once its upper bound is before ``cutoff``. The legacy partition holds every row from
+    before the swap and is catalogued as soon as it is attached, so that each of its sub-ranges becomes eligible on its
+    own age instead of waiting for the partition's upper bound, the swap time, to age past the window. Its drop still
+    waits for every sub-range. Units are taken in upper-bound order and skipped while they cannot move:
 
     - ``DROPPED`` units and ``BLOCKED`` statistics units.
     - ``DUMP_FAILED`` units whose last attempt is not before ``dump_retry_due_before``.
@@ -766,13 +769,16 @@ def select_next_unit(
     # Each entry sorts by (upper, lower); a partition without catalog rows sorts as its whole range.
     entries: list[tuple[tuple[datetime, datetime], CatalogUnit | PartitionBounds]] = []
     for partition in attached.values():
-        if partition.upper is None or partition.upper >= cutoff:
+        if partition.upper is None:
             continue
         partition_units = units_by_partition.get(partition.name)
         if not partition_units:
-            entries.append(((partition.upper, partition.lower or datetime.min), partition))
+            if partition.is_legacy or partition.upper < cutoff:
+                entries.append(((partition.upper, partition.lower or datetime.min), partition))
             continue
-        entries.extend((_unit_sort_key(partition_unit), partition_unit) for partition_unit in partition_units)
+        entries.extend(
+            (_unit_sort_key(partition_unit), partition_unit) for partition_unit in partition_units if partition_unit.upper < cutoff
+        )
     for partition_name, detached_units in units_by_partition.items():
         if partition_name in attached:
             continue
@@ -811,6 +817,32 @@ def select_next_unit(
     return NextUnit()
 
 
+def _eligible_uppers(
+    dataset: KudosRetentionDataset,
+    partitions: Sequence[PartitionBounds],
+    catalog: Sequence[CatalogUnit],
+    cutoff: datetime,
+) -> list[datetime]:
+    """Return the upper bounds past ``cutoff`` of the dataset's attached partitions, by sub-range for the legacy partition.
+
+    The legacy partition's units age individually (see ``select_next_unit``), so its backlog age is that of its oldest
+    sub-range still online, and the partition bound counts only while the partition has no catalog rows yet.
+    """
+    legacy_uppers_by_partition: dict[str, list[datetime]] = defaultdict(list)
+    for catalog_unit in catalog:
+        if catalog_unit.dataset == dataset and catalog_unit.state != KudosRetentionState.DROPPED:
+            legacy_uppers_by_partition[catalog_unit.partition_name].append(catalog_unit.upper)
+    uppers: list[datetime] = []
+    for partition in partitions:
+        if partition.dataset != dataset or partition.upper is None:
+            continue
+        if partition.is_legacy and partition.name in legacy_uppers_by_partition:
+            uppers.extend(upper for upper in legacy_uppers_by_partition[partition.name] if upper < cutoff)
+        elif partition.upper < cutoff:
+            uppers.append(partition.upper)
+    return uppers
+
+
 def _dump_retry_due(unit: CatalogUnit, due_before: datetime | None) -> bool:
     if unit.state != KudosRetentionState.DUMP_FAILED or due_before is None or unit.last_dump_attempt_at is None:
         return True
@@ -2757,14 +2789,11 @@ def retention_health(*, now: datetime | None = None, policy: KudosRetentionPolic
     if now is None:
         now = _utcnow()
     partitions = [] if SQLITE_MODE else discover_partitions()
+    catalog = load_catalog()
     oldest_eligible_age: dict[KudosRetentionDataset, float | None] = {}
     for dataset in KudosRetentionDataset:
         cutoff = now - policy.window_for(dataset)
-        eligible_uppers = [
-            partition.upper
-            for partition in partitions
-            if partition.dataset == dataset and partition.upper is not None and partition.upper < cutoff
-        ]
+        eligible_uppers = _eligible_uppers(dataset, partitions, catalog, cutoff)
         oldest_eligible_age[dataset] = (cutoff - min(eligible_uppers)).total_seconds() if eligible_uppers else None
     blocked_by_error = db.session.execute(
         select(KudosRetentionUnit.error, func.count(KudosRetentionUnit.id))
@@ -2785,7 +2814,6 @@ def retention_health(*, now: datetime | None = None, policy: KudosRetentionPolic
     for pending_dataset, count in pending_by_dataset:
         pending_units_by_dataset[KudosRetentionDataset(pending_dataset)] = int(count)
     last_update = db.session.execute(select(func.max(KudosRetentionUnit.updated))).scalar_one()
-    catalog = load_catalog()
     db.session.commit()
     return KudosRetentionHealth(
         oldest_eligible_age_seconds=oldest_eligible_age,

tests/unit/test_kudos_retention.py

@@ -287,6 +287,29 @@ def test_selection_lets_checkpointed_currency_wait_for_an_archive() -> None:
     assert with_archive.unit == catalog[0]
 
 
+def test_legacy_partition_is_catalogued_before_its_upper_bound_ages_past_the_window() -> None:
+    legacy = _partition(STATISTICS, "kudos_stat_events_p_legacy", None, CUTOFF + WEEK)
+    weekly = _partition(STATISTICS, "kudos_stat_events_p_20200203", CUTOFF, CUTOFF + WEEK)
+    selection = select_next_unit(dataset=STATISTICS, partitions=[weekly, legacy], catalog=[], cutoff=CUTOFF, archive_configured=False)
+    assert selection.partition_needing_units == legacy
+    assert select_next_unit(dataset=STATISTICS, partitions=[weekly], catalog=[], cutoff=CUTOFF, archive_configured=False).unit is None
+
+
+def test_legacy_sub_ranges_become_eligible_on_their_own_age() -> None:
+    legacy = _partition(STATISTICS, "kudos_stat_events_p_legacy", None, CUTOFF + WEEK)
+    old = _unit(1, STATISTICS, legacy.name, None, CUTOFF - WEEK, KudosRetentionState.PENDING)
+    young = _unit(2, STATISTICS, legacy.name, CUTOFF - WEEK, CUTOFF + WEEK, KudosRetentionState.PENDING)
+    assert _select(STATISTICS, [legacy], [old, young], CUTOFF).unit == old
+    # The sub-range that still receives rows waits, and so does the drop of the whole partition.
+    dumped = _unit(1, STATISTICS, legacy.name, None, CUTOFF - WEEK, KudosRetentionState.DUMPED)
+    assert _select(STATISTICS, [legacy], [dumped, young], CUTOFF).unit is None
+    assert _select(STATISTICS, [legacy], [dumped, young], CUTOFF + 2 * WEEK).unit == young
+
+
+def _select(dataset: KudosRetentionDataset, partitions: list[PartitionBounds], catalog: list[Any], cutoff: datetime) -> Any:
+    return select_next_unit(dataset=dataset, partitions=partitions, catalog=catalog, cutoff=cutoff, archive_configured=False)
+
+
 def test_selection_drops_the_legacy_partition_only_when_every_sub_range_is_ready() -> None:
     legacy = _partition(STATISTICS, "kudos_stat_events_p_legacy", None, MONDAY + 2 * WEEK)
     dumped = _unit(1, STATISTICS, legacy.name, None, MONDAY + WEEK, KudosRetentionState.DUMPED)