From 7337c844f16a22aed1f02d3c4bfcc07f29f4eb25 Mon Sep 17 00:00:00 2001 From: tazlin Date: Wed, 4 Nov 2026 00:24:25 -0500 Subject: [PATCH] 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. --- ...rchive-currency-after-the-online-window.md | 6 +-- docs/explanation/kudos_retention.md | 5 +- docs/how-to/kudos_ledger_operations.md | 23 ++++---- docs/reference/kudos_accounting.md | 12 +++-- horde/database/kudos_retention.py | 52 ++++++++++++++----- tests/unit/test_kudos_retention.py | 23 ++++++++ 6 files changed, 90 insertions(+), 31 deletions(-) diff --git a/docs/decisions/0042-roll-up-statistics-and-archive-currency-after-the-online-window.md b/docs/decisions/0042-roll-up-statistics-and-archive-currency-after-the-online-window.md index fb96440c..52a36085 100644 --- a/docs/decisions/0042-roll-up-statistics-and-archive-currency-after-the-online-window.md +++ b/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. diff --git a/docs/explanation/kudos_retention.md b/docs/explanation/kudos_retention.md index 8be68da9..26a0d54f 100644 --- a/docs/explanation/kudos_retention.md +++ b/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. diff --git a/docs/how-to/kudos_ledger_operations.md b/docs/how-to/kudos_ledger_operations.md index 000950d3..02ffbedb 100644 --- a/docs/how-to/kudos_ledger_operations.md +++ b/docs/how-to/kudos_ledger_operations.md @@ -317,14 +317,17 @@ docker exec -w /app -e PYTHONPATH=/app 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 units of partition _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 units of partition
_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: diff --git a/docs/reference/kudos_accounting.md b/docs/reference/kudos_accounting.md index f2097e4f..5a154560 100644 --- a/docs/reference/kudos_accounting.md +++ b/docs/reference/kudos_accounting.md @@ -439,9 +439,11 @@ A unit's `source_range` is `/_`. 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 | diff --git a/horde/database/kudos_retention.py b/horde/database/kudos_retention.py index 679921a0..5fb7cad5 100644 --- a/horde/database/kudos_retention.py +++ b/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, diff --git a/tests/unit/test_kudos_retention.py b/tests/unit/test_kudos_retention.py index e45430c2..d202c89d 100644 --- a/tests/unit/test_kudos_retention.py +++ b/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)