From 4580ec70a20dab0efed39dcb0c7de8796042976c Mon Sep 17 00:00:00 2001 From: tazlin Date: Tue, 3 Nov 2026 23:12:54 -0500 Subject: [PATCH] fix(kudos): the ledger swap holds the applier lock and bounds its lock waits, and retention guards lost statistics dumps and restored partitions - sql_statements/5.1.16.txt: the kudos_ledger swap takes the applier advisory lock (KUDOS_APPLIER_LOCK) before it locks users. The applier claims ledger rows and then updates users, the reverse of the swap's order. The applier only tries that lock, so it skips its cycles until the swap commits and does not deadlock with it. - After the advisory lock, each lock wait of the swap ends after 500 ms, below deadlock_timeout. A request that inserts a posting before it updates users can still form a lock cycle with the swap. The swap then fails with lock_not_available before the deadlock check runs in the request, so the request continues and the file is rerun. The file stops with an error when deadlock_timeout is 500 ms or less. - archive evict runs evict_statistics_dumps. It keeps the dumps of statistics units that are not dropped, because their raw rows are still online. It sets the object_key of each evicted unit to NULL, also after a failed batch. The command exits 1 when the kept dumps hold the tier above the budget. KudosArchive.enforce_budget takes kept_keys for this. - The tick does not clear the key of a missing statistics dump without a trace. A dumped unit whose object is missing goes to dump_failed and is dumped again on the next tick. A detached one blocks with archive object missing. - A dropped statistics unit keeps the key of a missing object, so archive audit reports the loss on every run. - _check_statistics_dumps runs with the currency check before the detach and before the drop. The raw rows of a unit stay online when its dump was removed after the last listing. - archive object missing is waivable for statistics. retention unblock --waive-dump refuses a unit of a detached partition, so the drop after a waiver starts from an attached partition. - A dataset with an attached partition that has dropped catalog units does not make a transition, and the tick returns blocked with RESTORED_PARTITION_REFUSAL. Such a partition was restored under its original name, and the catalog does not describe its rows. retention_health lists restored_partitions, and the gauge horde.kudos.retention.restored_partitions exports the count. - Tests: the migration tests run the swap against a concurrent applier cycle and a concurrent request, and against a short deadlock_timeout. The retention tests cover a vanished dump in each state and an eviction that keeps online dumps. A further test restores a dropped partition under its name. The reconcile test expects dump_failed for a vanished dump, and the rehearsal grace test expects the restored-partition block after a future reference time drops the current week. - Docs: the how-to's swap, eviction, missing object, rehearsal and rollback text, the reference's block reason, eviction, reconcile, metric, log and health tables, and the explanation's diagnosis table. --- docs/explanation/kudos_retention.md | 3 +- docs/how-to/kudos_ledger_operations.md | 78 ++++-- docs/reference/kudos_accounting.md | 58 +++-- horde/database/kudos_retention.py | 232 ++++++++++++++++-- horde/database/threads.py | 2 + horde/kudos_archive.py | 9 +- horde/metrics.py | 5 + sql_statements/5.1.16.txt | 24 +- .../test_kudos_retention_migration.py | 158 +++++++++++- tests/unit/test_kudos_retention.py | 190 +++++++++++++- tools/kudos_ledger_admin.py | 20 +- 11 files changed, 687 insertions(+), 92 deletions(-) diff --git a/docs/explanation/kudos_retention.md b/docs/explanation/kudos_retention.md index 91e804cf..8be68da9 100644 --- a/docs/explanation/kudos_retention.md +++ b/docs/explanation/kudos_retention.md @@ -194,7 +194,8 @@ It costs these: | Blocked, `currency tier would exceed budget` | The currency export would pass the currency budget. Currency retention stops and statistics continue | [Handling a blocked unit](../how-to/kudos_ledger_operations.md#handling-a-blocked-unit) | | Blocked, `statistics tier would exceed budget` | The statistics dump would pass the statistics budget. Nothing is uploaded, and later statistics units continue | [Handling a blocked unit](../how-to/kudos_ledger_operations.md#handling-a-blocked-unit) | | Blocked, `archive mismatch` | The object read back disagrees with its manifest | [Handling a blocked unit](../how-to/kudos_ledger_operations.md#handling-a-blocked-unit) | -| Blocked, `archive object missing` | A currency unit past its export has lost its archive object, so it is not dropped | [Handling a blocked unit](../how-to/kudos_ledger_operations.md#handling-a-blocked-unit) | +| Blocked, `archive object missing` | A currency unit past its export, or a dumped statistics unit before its drop, has lost its archive object, so it is not dropped | [Handling a blocked unit](../how-to/kudos_ledger_operations.md#handling-a-blocked-unit) | +| Restored partitions | A dropped partition was attached again under its original name. Its dataset does not make a transition until the partition is detached | [Rolling back a removed partition](../how-to/kudos_ledger_operations.md#rolling-back-a-removed-partition) | | Blocked, `online drift since export` or `late rows` | Rows in the unit changed after its replacement recorded them | [Recovering from a late-row or chain block](../how-to/kudos_ledger_operations.md#recovering-from-a-late-row-or-chain-block) | | `dump_failed`, or blocked with `dump retries exhausted` | Statistics dumps are failing, and the rollup is already committed | [Handling a blocked unit](../how-to/kudos_ledger_operations.md#handling-a-blocked-unit) | | Held partitions | `KUDOS_RETENTION_HOLD_DROPS` keeps drop-ready partitions attached. Every other step runs | [Enabling retention](../how-to/kudos_ledger_operations.md#enabling-retention) | diff --git a/docs/how-to/kudos_ledger_operations.md b/docs/how-to/kudos_ledger_operations.md index 2b4dfbd3..000950d3 100644 --- a/docs/how-to/kudos_ledger_operations.md +++ b/docs/how-to/kudos_ledger_operations.md @@ -162,7 +162,8 @@ with [restoring an archived partition](#restoring-an-archived-partition-into-a-s Preconditions: the previous schema is applied, the application release that maps the partitioned tables is not yet deployed, the `psql` client can reach the primary, and no long-running dump or backup (`pg_dump`, a base backup) is running or scheduled to start during the run. A dump holds `ACCESS SHARE` on every table for its whole duration, so -the swap cannot take its exclusive lock within the 5 second `lock_timeout` and fails. End state: both event tables +the swap cannot take its exclusive lock within its `lock_timeout` and fails. The `kudos_ledger` swap also refuses to +start when the server's `deadlock_timeout` is 500 ms or less. End state: both event tables are partitioned, each with a `_p_legacy` partition and eight weekly partitions from `swap_ts`, and the retention tables exist and are empty except for the backfilled keys. @@ -203,11 +204,14 @@ the backfilled keys. first UTC Monday 00:00 at least 24 hours away. It then adds `
_legacy_range CHECK (created < swap_ts)` and validates it with one scan that does not block writes. The swap transaction renames the table to `
_p_legacy`, creates the partitioned parent, and attaches the old table and its indexes without a scan or - copy. It holds an exclusive lock for a short time. The `kudos_ledger` swap first locks `users` in - `SHARE ROW EXCLUSIVE` mode, the order application transactions take the two locks in, so the swap cannot deadlock - with them. Writes to `users` wait until the swap commits. The swap fails if it cannot take a lock within the - 5 second `lock_timeout`, and then the file is rerun as described in - [rerunning after a failure](#rerunning-after-a-failure). + copy. It holds an exclusive lock for a short time. The `kudos_ledger` swap first takes the applier's advisory + lock, so the applier skips its cycles until the swap commits. It then locks `users` in `SHARE ROW EXCLUSIVE` mode + and `kudos_ledger` right after. Writes to `users` wait until the swap commits. A request that inserted a posting + and then updates `users` can still form a lock cycle with the swap. Each lock wait of the `kudos_ledger` swap + ends after 500 ms, below `deadlock_timeout`, so in that case the swap fails with `lock_not_available` and the + request continues. The `kudos_stat_events` swap waits up to 5 seconds for each lock. After a lock failure, rerun + the file as described in [rerunning after a failure](#rerunning-after-a-failure). On a busy primary the + `kudos_ledger` swap can need several reruns. The swap window runs from the moment `swap_ts` is set until `swap_ts`. The CHECK rejects new rows with `created` at or after `swap_ts` until the swap commits, so the window must outlast the validation and the swap, @@ -381,6 +385,21 @@ units stay `blocked` until an operator releases them. A partition that was already dropped cannot be reattached. Its currency rows can still be read with [restoring an archived partition](#restoring-an-archived-partition-into-a-scratch-table). +A dropped partition restored from a backup and attached under its original name keeps its `dropped` catalog units. +The tick then does not make a transition in that dataset. It returns `"outcome": "blocked"` and logs +`Kudos retention blocked dataset: an attached partition has dropped catalog units: `, +`retention status` lists the partition in `restored_partitions`, and `horde.kudos.retention.restored_partitions` is +above 0. Read the rows you need, then detach the table again: + +```sql +ALTER TABLE DETACH PARTITION ; +``` + +The checkpoints, rollups and archive objects of the dropped units still describe the range, so keep the detached +table outside the parent or drop it. Do not delete or edit the catalog rows to make the table eligible again. A +currency unit recomputed over a recorded boundary blocks with `boundary already checkpointed`. Verify: +`retention status` reports an empty `restored_partitions`, and the next tick steps or reports `idle`. + ## 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 @@ -401,6 +420,11 @@ makes the partitions that still receive rows eligible, so without `--confirm-liv database it drops the current week's rows. After the rehearsal, `retention status` reports `"partition_coverage_ok": true` once partitions for the real current time exist again. +Partition maintenance creates a dropped current week again under its weekly name. That table has `dropped` catalog +units, so once it is attached the dataset stops with `an attached partition has dropped catalog units`, and +`retention status` lists it in `restored_partitions`. To continue the rehearsal past that point, detach and drop the +table as described in [rolling back a removed partition](#rolling-back-a-removed-partition). + The drop grace is measured with the tick's reference time. With `KUDOS_RETENTION_DROP_GRACE_DAYS` set, a rehearsal tick detaches a partition with `detached_at` set to its `--as-of`, and the partition is dropped only by a later tick whose `--as-of` is at least the grace past it. Advance `--as-of` past `drop_due_at` from `retention status` to see the @@ -451,25 +475,34 @@ fixed, the unit is back in `pending` or `checkpointed`, and it advances on later - `statistics tier would exceed budget`: the dump would push the statistics tier past `KUDOS_ARCHIVE_STATISTICS_BUDGET_BYTES`, so nothing was uploaded. Later statistics units continue, and each of them blocks the same way until there is room. Raise the budget and restart, or free space by evicting the oldest - dumps: + dumps of dropped units: ```sh uv run python tools/kudos_ledger_admin.py archive evict --tier statistics --budget-bytes ``` - The command prints the deleted keys. When a delete fails it prints the keys deleted before the failure and exits - 1. On a bucket with versioning, a deleted object stays stored as a noncurrent version that the tier listing does - not count, so free the space by the deployment's own procedure as well. The next tick clears the catalog keys of - every deleted object. Then unblock the unit, which rolls it up again - and retries the dump. Verify: the unit reaches `dumped` and its `object_key` is set. To drop the unit without a - dump instead, run `retention unblock --waive-dump`. - - `archive object missing`: a currency unit past its export has no archive object. Either the tick's archive - listing does not hold the unit's key, or the drop found no key. Nothing was dropped, and the rows are still - online or in the detached table. Find out why the object is gone before going on. For a unit in an attached - partition, unblock it: it returns to `checkpointed`, and the next ticks export, verify and drop it again from - the online rows. For a unit of a detached partition, `retention unblock` refuses, so run - `retention reattach ` first and then unblock. Verify: the unit reaches `verified` with a new - `object_key`, and `archive audit --tier currency` reports no `missing_object` line for it. + The command prints the deleted keys and `tier_bytes`, and sets the catalog key of each evicted unit to NULL. The + dumps of units not yet dropped are kept, because their raw rows are still online. When those dumps alone keep + the tier above the budget, the command exits 1 with an `error`. When a delete fails it prints the keys deleted + before the failure and exits 1. On a bucket with versioning, a deleted object stays stored as a noncurrent + version that the tier listing does not count, so free the space by the deployment's own procedure as well. Then + unblock the unit, which rolls it up again and retries the dump. Verify: the unit reaches `dumped` and its + `object_key` is set. To drop the unit without a dump instead, run `retention unblock --waive-dump`. + - `archive object missing`: the archive object of a currency unit past its export, or of a dumped statistics unit + before its drop, is missing. Either the tick's archive listing does not hold the unit's key, or the drop found + that the key or the object is missing. Eviction does not reach these units, so the object was removed some + other way. Nothing was dropped, and the rows are still online or in the detached table. Find out why the object is gone before going on. On a + bucket with versioning, the object may still be stored as a noncurrent version. For a unit of a detached + partition, `retention unblock` refuses, so run `retention reattach ` first. Then unblock the unit. A + currency unit returns to `checkpointed`, and the next ticks export, verify and drop it again from the online + rows. A statistics unit returns to `pending`, and the next ticks roll it up, dump it and drop it again. To drop + a statistics unit without a dump instead, run `retention unblock --waive-dump` after the reattach. + Verify: the unit reaches `verified` or `dumped` with an `object_key`, and `archive audit` reports no + `missing_object` line for it. + + A `dumped` statistics unit in an attached partition whose dump is missing from the listing does not block. The + tick moves it to `dump_failed` and logs `Kudos retention dump failed for : the archive does not hold + , so the unit is dumped again`. The next tick dumps the rows again, and the usual dump retries apply. - `archive mismatch`: the object read back does not match its manifest. The object is kept. The warning includes the mismatched fields. Check the object store, and restore the object into a scratch table if the rows need inspection. @@ -577,8 +610,9 @@ object are in a scratch table in the application database. Qualify the scratch t FROM kudos_retention_catalog WHERE object_key IS NOT NULL ORDER BY unit_upper; ``` - A statistics object deleted by `archive evict`, or lost in any other way, leaves its unit with a null - `object_key` once a tick has run. A statistics unit in `dump_failed` or `dump_waived` does not have an object. + A statistics object deleted by `archive evict` leaves its unit with a null `object_key`. A dropped statistics + unit whose object was lost in any other way keeps the key, and `archive audit` reports it as a `missing_object` + line. A statistics unit in `dump_failed` or `dump_waived` does not have an object. 2. Restore it into a table name that does not exist yet: ```sh diff --git a/docs/reference/kudos_accounting.md b/docs/reference/kudos_accounting.md index 5f075150..bb7bc87b 100644 --- a/docs/reference/kudos_accounting.md +++ b/docs/reference/kudos_accounting.md @@ -475,7 +475,8 @@ Otherwise it returns a blocked currency unit whose checkpoint is committed (`row_count` is set) to `checkpointed`, so export and verification run again against that checkpoint. Every other blocked unit returns to `pending`, and its steps run again from the start. Each step replaces what an earlier attempt wrote. `retention unblock --waive-dump` moves a `dump_failed` statistics unit, or one blocked for -`dump retries exhausted` or `statistics tier would exceed budget`, to `dump_waived`. +`dump retries exhausted`, `statistics tier would exceed budget` or `archive object missing`, to `dump_waived`. It +refuses a unit of a detached partition until the partition is reattached. `retention unblock --recheckpoint` returns a blocked currency unit and every later currency unit to `pending`, and clears their `error`, `row_count`, `max_id` and `object_key`. In the same transaction it deletes the @@ -567,7 +568,7 @@ The `error` of a blocked unit is one of the `RetentionBlockReason` values. | `online drift since export` | `currency`, at `exported` | The manifest's row count differs from `row_count`, or the row count, amount sum or per-user totals of the online rows differ from the archived object | | `late rows` | Both, at the drop | A unit's row count or highest `id` differs from `row_count` and `max_id`, or the unit has no recorded count | | `dump retries exhausted` | `statistics`, at `dump_failed` | A dump retry failed more than `KUDOS_RETENTION_DUMP_RETRY_DAYS` after the first failure | -| `archive object missing` | `currency`, at `exported`, `verified` or `detached`, or at the drop | The tick's archive listing does not hold the unit's `object_key`, or the drop found a currency unit with no `object_key`. `retention unblock` returns the unit to `checkpointed`, so it is exported and verified again. A unit of a detached partition is reattached first | +| `archive object missing` | `currency`, at `exported`, `verified` or `detached`, or at the drop. `statistics`, at `detached` with a `dumped` state before the detach, or at the drop | The tick's archive listing does not hold the unit's `object_key`, or the drop found a currency unit or a dumped statistics unit whose `object_key` or object is missing. A unit of a detached partition is reattached first. `retention unblock` then returns a currency unit to `checkpointed`, so it is exported and verified again, and a statistics unit to `pending`, so it is rolled up and dumped again. `retention unblock --waive-dump` drops a statistics unit without the dump | | `reattached by operator` | Both, at `detached` | `retention reattach` attached the unit's partition again. `retention unblock` returns the unit to `state_before_detach` | ### Reconciliation after retention @@ -670,24 +671,33 @@ multipart upload and blocks the unit with the same reason without replacing an o count completed data objects and exclude manifests and incomplete multipart uploads. Every writer to a tier holds the retention lock, including `archive evict`. -`archive evict --tier statistics --budget-bytes ` runs `KudosArchive.enforce_budget`, which deletes the oldest -statistics objects and their manifests, ordered by unit upper bound, until the statistics tier fits ``. Currency -objects are never deleted. The deletes go out one batch at a time. When one fails, `ArchiveEvictionError` holds the -keys deleted before it, and the command prints them. +`archive evict --tier statistics --budget-bytes ` runs `evict_statistics_dumps`, which deletes the oldest +statistics objects and their manifests, ordered by unit upper bound, until the statistics tier fits ``. The dump of +a statistics unit that is not `dropped` is kept, because its raw rows are still online. Objects that are not recorded by any +unit are deleted like dropped dumps. Currency objects are never deleted. The deletes go out one batch at a time. When one +fails, `ArchiveEvictionError` holds the keys deleted before it, and the command prints them. The command sets the +`object_key` of each evicted unit to NULL, also after a failed batch, so an evicted dump is not reported as lost. -With an archive configured, every tick lists both tiers after its step, while it still holds the retention lock. -A unit whose `object_key` is not in the listing gets a NULL `object_key`, and the tick logs -`Kudos retention cleared object key of unit (, )`. The unit keeps its -state, except a currency unit in `exported`, `verified` or `detached`. That unit becomes `blocked` with -`archive object missing`, and the line starts with `Kudos retention blocked` instead, because a drop would remove -the only copy of its postings. Before the detach and again immediately before the drop, currency retention checks -every currency unit of the partition against the archive: the object exists, its manifest is readable, the stored -size equals the manifest's `compressed_bytes`, and the manifest's `row_count` equals the unit's recorded count. A -missing key or object blocks the unit with `archive object missing`, a mismatch blocks it with `archive mismatch`, -and an unreachable store or unreadable manifest fails the tick so that a later tick retries. The check compares -metadata only. The digest was verified at the `verified` step, and `archive audit` re-reads objects. The post-step -listing is housekeeping and does not gate the drop. The tier sizes from the same listing are the -`horde.kudos.archive.tier_bytes` values. +With an archive configured, every tick lists both tiers after its step, while it still holds the retention lock. A +unit whose `object_key` is not in the listing was removed outside the horde, or by an eviction that stopped before it +updated the catalog. What the tick does depends on the unit: + +| Unit | Change | Log line | +| --- | --- | --- | +| `currency` in `exported`, `verified` or `detached` | `blocked` with `archive object missing`, `object_key` NULL, because a drop would remove the only copy of its postings | `Kudos retention blocked unit : archive object missing: ...` | +| `statistics` in `dumped` | `dump_failed`, `object_key` NULL, `error` `archive object missing: `. The dump retry window starts at that tick, and the next tick dumps the rows again | `Kudos retention dump failed for : the archive does not hold , so the unit is dumped again` | +| `statistics` in `detached`, with `dumped` before the detach | `blocked` with `archive object missing`, `object_key` NULL | `Kudos retention blocked unit : archive object missing: ...` | +| `statistics` in `dropped` | None. The unit keeps its `object_key`, so `archive audit` reports the lost object on every run | None | +| Any other | `object_key` NULL | `Kudos retention cleared object key of unit (, )` | + +Before the detach and again immediately before the drop, retention checks every unit of the partition against the +archive. For a currency unit, the object exists, its manifest is readable, the stored size equals the manifest's +`compressed_bytes`, and the manifest's `row_count` equals the unit's recorded count. For a statistics unit in +`dumped`, or `detached` with `dumped` before the detach, the object exists. A missing key or object blocks the unit +with `archive object missing`, a currency mismatch blocks it with `archive mismatch`, and an unreachable store or +unreadable manifest fails the tick so that a later tick retries. The check compares metadata only. The digest was +verified at the `verified` step, and `archive audit` re-reads objects. The post-step listing does not gate the drop. +The tier sizes from the same listing are the `horde.kudos.archive.tier_bytes` values. ### Retention metrics and log lines @@ -701,6 +711,7 @@ listing is housekeeping and does not gate the drop. The tier sizes from the same | `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` | +| `horde.kudos.retention.restored_partitions` | Gauge, `{partition}` | None | The number of `restored_partitions` from `retention_health`. Above 0, the tick does not make a transition in the dataset of each such partition | | `horde.kudos.archive.tier_bytes` | Gauge, `By` | `horde.kudos.archive.tier` | Stored bytes of each tier, excluding manifests, from the tick's listing. Not recorded without an archive, and kept at the last value when the listing fails | | `horde.kudos.archive.budget_bytes` | Gauge, `By` | `horde.kudos.archive.tier` | One series per tier that has a budget: `currency` from `KUDOS_ARCHIVE_BUDGET_BYTES`, `statistics` from `KUDOS_ARCHIVE_STATISTICS_BUDGET_BYTES`. Not recorded without an archive | @@ -709,8 +720,8 @@ Two warnings are under log contract. Log-based alerting matches their prefixes, | Prefix | Constant | Logged when | Full line | | --- | --- | --- | --- | -| `Kudos retention blocked` | `RETENTION_BLOCKED_LOG_PREFIX` | A unit becomes `blocked` | `Kudos retention blocked unit : `, optionally followed by `: ` | -| `Kudos retention dump failed` | `RETENTION_DUMP_FAILED_LOG_PREFIX` | A statistics dump attempt fails and the unit stays `dump_failed`, or the dump step waives a unit because no archive is configured | `Kudos retention dump failed for : ( consecutive failures since )`, or `...: archive not configured` | +| `Kudos retention blocked` | `RETENTION_BLOCKED_LOG_PREFIX` | A unit becomes `blocked`, or a tick does not make a transition in a dataset because of a restored partition | `Kudos retention blocked unit : `, optionally followed by `: `, or `Kudos retention blocked dataset: an attached partition has dropped catalog units: ` | +| `Kudos retention dump failed` | `RETENTION_DUMP_FAILED_LOG_PREFIX` | A statistics dump attempt fails and the unit stays `dump_failed`, the dump step waives a unit because no archive is configured, or the tick finds the dump of a `dumped` unit missing | `Kudos retention dump failed for : ( consecutive failures since )`, `...: archive not configured`, or `...: the archive does not hold , so the unit is dumped again` | The dump-failed warning is logged on every failed attempt. The attempt that exhausts the retry window logs the blocked warning instead, with `dump retries exhausted` and the same detail. Without an archive, the dump-failed warning is @@ -731,6 +742,7 @@ does not count it. Other retention log lines are not under contract. | `held_partitions` | Partitions whose units, other than `dropped` ones, are all drop-ready, while `hold_drops` is true. 0 otherwise | | `drop_grace_days` | The drop grace of the running policy in days, from `KUDOS_RETENTION_DROP_GRACE_DAYS` or its default | | `detached_partitions` | One entry per partition with `detached` units, oldest detach first: `dataset`, `partition_name`, `detached_at` and `drop_due_at` (`detached_at` plus the grace) | +| `restored_partitions` | One entry per attached partition with `dropped` catalog units, as `dataset` and `partition_name`. Such a partition was restored under its original name. The dataset does not make a transition until the partition is detached | ### Retention commands @@ -740,11 +752,11 @@ These are subcommands of `tools/kudos_ledger_admin.py`. Each prints JSON. | --- | --- | | `retention status` | Prints the health fields, `archive_configured`, `archive_tier_bytes`, and every unit not yet dropped with its catalog columns and `source_range`. Read-only | | `retention step [--dataset ] [--as-of ] [--confirm-live-data-loss]` | Runs one tick as the scheduler does. With `--dataset currency` or `--dataset statistics`, only that dataset's units are stepped. `--as-of` is a rehearsal control: every eligibility, key-expiry and partition-creation decision is taken as of that UTC time, which must be no earlier than the current time and no later than the current time plus the longer window, the drop grace and four partition widths, and retention must be enabled. With a drop grace, a partition detached at one `--as-of` is dropped only by a later tick whose `--as-of` is at least the grace past it. A time later than now is refused without `--confirm-live-data-loss`, because against a live database it removes current data. Use that flag on a restored copy only. It can drop a partition | -| `retention unblock [--waive-dump \| --recheckpoint]` | Returns a blocked unit to `state_before_detach` when it is set (a unit blocked by `retention reattach`), else to `checkpointed` (currency unit with a committed checkpoint) or `pending`, and clears `error`. With `--waive-dump`, moves a `dump_failed` statistics unit, or one blocked for `dump retries exhausted` or `statistics tier would exceed budget`, to `dump_waived`. With `--recheckpoint`, returns a blocked currency unit and every later currency unit to `pending` so their checkpoints are recomputed, and refuses when a later currency unit is `dropped` or `detached`. Refuses a unit of a `detached` partition. Exits with an error for any other unit, or while a tick holds the retention lock | +| `retention unblock [--waive-dump \| --recheckpoint]` | Returns a blocked unit to `state_before_detach` when it is set (a unit blocked by `retention reattach`), else to `checkpointed` (currency unit with a committed checkpoint) or `pending`, and clears `error`. With `--waive-dump`, moves a `dump_failed` statistics unit, or one blocked for `dump retries exhausted`, `statistics tier would exceed budget` or `archive object missing`, to `dump_waived`. With `--recheckpoint`, returns a blocked currency unit and every later currency unit to `pending` so their checkpoints are recomputed, and refuses when a later currency unit is `dropped` or `detached`. Refuses a unit of a `detached` partition. Exits with an error for any other unit, or while a tick holds the retention lock | | `retention reattach ` | Attaches the `detached` partition of the unit back to its parent over its recorded range and blocks its units with `reattached by operator`, keeping `state_before_detach`. `retention unblock` on each unit returns it to that state. Refuses when the partition is not `detached` or its table no longer exists. Exits with an error while a tick holds the retention lock | | `statement --from --to ` | Prints the user's lines from `kudos_ledger_daily` for those UTC days, summed across `source_range`. Days still online do not have lines yet | | `archive restore --into [--append]` | Copies an archive object into a new scratch table, `table` or `schema.table`, created from the manifest's `column_types` alone, without reference to the live event tables. A version 1 manifest does not record column types and is refused. An existing table is refused, because a repeated restore would hold every row twice and each restore verifies only its own rows. With `--append` the rows are added to an existing table, and the output has `"appended": true` and a note. Requires the archive configuration | -| `archive evict --tier statistics --budget-bytes ` | Deletes the oldest statistics objects until the statistics tier fits `` and prints the deleted keys. On a failed delete it prints the keys deleted before it with the error and exits 1. The next tick clears the catalog keys of the deleted objects | +| `archive evict --tier statistics --budget-bytes ` | Deletes the oldest statistics objects of `dropped` units and objects without a unit until the statistics tier fits ``, sets the `object_key` of each evicted unit to NULL, and prints the deleted keys and `tier_bytes`, the stored bytes left. Exits 1 with an `error` when the dumps of units not yet dropped keep the tier above ``. On a failed delete it prints the keys deleted before it with the error and exits 1 | | `archive audit [--tier currency\|statistics] [--since-key ]` | Reads every object of the tier, both tiers by default, back against its manifest: digest, stored bytes, row count, amount sum, columns and per-user totals. Prints one JSON line per object (`"kind": "object"`, `matches_manifest`, `problems`), one per catalog unit whose object is missing (`"kind": "missing_object"`), then a summary line with `objects`, `problems`, `last_key`, `noncurrent_bytes` and `delete_markers`. The last two map each audited tier to the summed size of the noncurrent object versions under its prefix, manifests included, and to its count of delete markers. They show storage a versioned bucket still holds for evicted or replaced objects, which the `horde.kudos.archive.tier_bytes` gauge and `retention status` do not count. A tier's value is `null` when the store did not list its versions, which differs from `0`. Neither figure counts as a problem. A unit is missing its object when its `object_key` is not listed, or when it is a currency unit in `exported`, `verified`, `detached` or `dropped` without an `object_key`. Exits 1 on any problem, including an unreadable manifest. `--since-key` audits only objects whose key sorts after ``, to resume from a summary's `last_key`. It reads the catalog and changes nothing | ### Privacy disclosure diff --git a/horde/database/kudos_retention.py b/horde/database/kudos_retention.py index d06d2eb0..c931a00e 100644 --- a/horde/database/kudos_retention.py +++ b/horde/database/kudos_retention.py @@ -75,6 +75,7 @@ from horde.kudos_archive import ( ARCHIVE_REQUIRED_ENV, ArchiveBudgetExceeded, ArchiveColumn, + ArchiveEvictionError, ArchiveObjectAudit, ArchiveTier, KudosArchive, @@ -126,6 +127,10 @@ DUMP_RETRY_SPACING: Final[timedelta] = timedelta(minutes=15) STALE_UPLOAD_AGE: Final[timedelta] = timedelta(days=1) """How long an incomplete multipart upload stays open before a tick aborts it. It is far longer than any export.""" FUTURE_REFERENCE_REFUSAL: Final[str] = "reference time is in the future. Pass --confirm-live-data-loss on a restored copy only" +RESTORED_PARTITION_REFUSAL: Final[str] = "an attached partition has dropped catalog units" +"""The detail of a ``BLOCKED`` tick result for a dataset with an attached partition whose catalog units are ``DROPPED``. +Such a partition was restored from a backup under its original name. The catalog does not describe its rows, so the +dataset does not make a transition until an operator detaches the partition.""" """The detail of a tick refused because its reference time is later than the real clock.""" DEFAULT_WINDOW_DAYS: Final[int] = 30 @@ -195,16 +200,23 @@ class RetentionBlockReason(enum.StrEnum): """Statistics dumps kept failing for longer than the retry window. ``retention unblock --waive-dump`` accepts dropping the unit without a dump.""" ARCHIVE_OBJECT_MISSING = "archive object missing" - """A currency unit past its export has no archive object: the tick's archive listing does not hold its key, or the - drop found no key. ``retention unblock`` returns it to ``CHECKPOINTED``, so it is exported and verified again from - the rows still online, and a detached partition is reattached first.""" + """The archive object of a currency unit past its export, or of a dumped statistics unit before its drop, is + missing. The tick's archive listing does not hold its key, or the drop found that the key or the object is + missing. A detached partition is reattached first. ``retention unblock`` then returns a currency unit to + ``CHECKPOINTED``, so it is exported and verified again from the rows still online, and a statistics unit to + ``PENDING``, so it is rolled up and dumped again. ``retention unblock --waive-dump`` drops a statistics unit + without the dump.""" REATTACHED = "reattached by operator" """An operator reattached the unit's detached partition. The unit stays blocked, so the partition is not detached again until an operator unblocks it, which restores ``state_before_detach``.""" WAIVABLE_DUMP_BLOCK_REASONS: Final[frozenset[str]] = frozenset( - {RetentionBlockReason.DUMP_RETRIES_EXHAUSTED, RetentionBlockReason.STATISTICS_BUDGET_EXCEEDED}, + { + RetentionBlockReason.DUMP_RETRIES_EXHAUSTED, + RetentionBlockReason.STATISTICS_BUDGET_EXCEEDED, + RetentionBlockReason.ARCHIVE_OBJECT_MISSING, + }, ) """The block reasons of a statistics unit whose dump an operator may waive, dropping its raw rows without a copy.""" @@ -548,6 +560,15 @@ class KudosRetentionHealth(TypedDict): """The policy's drop grace in days, so a deploy can check the reattach window the running policy keeps.""" detached_partitions: list[DetachedPartitionHealth] """Partitions whose units are ``DETACHED``, oldest detach first.""" + restored_partitions: list[RestoredPartitionHealth] + """Attached partitions with ``DROPPED`` catalog units. Their dataset does not make a transition while one is listed.""" + + +class RestoredPartitionHealth(TypedDict): + """Represents one attached partition whose catalog units are ``DROPPED``.""" + + dataset: KudosRetentionDataset + partition_name: str class DetachedPartitionHealth(TypedDict): @@ -832,6 +853,25 @@ def drop_ready_partition_count(catalog: Sequence[CatalogUnit]) -> int: return sum(1 for partition_units in _units_by_partition(catalog).values() if _partition_drop_ready(partition_units)) +def restored_partitions(partitions: Sequence[PartitionBounds], catalog: Sequence[CatalogUnit]) -> list[RestoredPartitionHealth]: + """Return the attached partitions with a ``DROPPED`` catalog unit, in dataset and name order. + + A dropped partition that is restored and attached under its original name keeps its catalog rows, and the + selection skips ``DROPPED`` units, so without this check the partition would stay attached and unaccounted. + + Args: + partitions: Every attached partition of both datasets. + catalog: Catalog rows of both datasets. + """ + attached = {(partition.dataset, partition.name) for partition in partitions} + dropped = { + (catalog_unit.dataset, catalog_unit.partition_name) for catalog_unit in catalog if catalog_unit.state == KudosRetentionState.DROPPED + } + return [ + RestoredPartitionHealth(dataset=dataset, partition_name=partition_name) for dataset, partition_name in sorted(attached & dropped) + ] + + def detached_partitions(catalog: Sequence[CatalogUnit], *, drop_grace: timedelta) -> list[DetachedPartitionHealth]: """Return the partitions kept detached for the drop grace period, with when each drop falls due, oldest detach first. @@ -1635,6 +1675,48 @@ def _check_currency_archive(context: _StepContext, siblings: Sequence[CatalogUni return None +def _holds_statistics_dump(unit: CatalogUnit) -> bool: + """Return whether ``unit`` is a statistics unit whose drop relies on its archive dump.""" + if unit.dataset != KudosRetentionDataset.STATISTICS: + return False + if unit.state == KudosRetentionState.DETACHED: + return unit.state_before_detach == KudosRetentionState.DUMPED + return unit.state == KudosRetentionState.DUMPED + + +def _check_statistics_dumps(context: _StepContext, siblings: Sequence[CatalogUnit]) -> RetentionTickResult | None: + """Return the outcome that stops the drop when a dumped statistics unit's object is missing or unreachable, else None. + + A dump can disappear after it was written, by a delete outside the horde. Eviction does not reach the dump of a + unit that is not ``DROPPED``. A missing dump blocks the unit with ``ARCHIVE_OBJECT_MISSING``, so its raw rows are + not dropped without a copy unless an operator waives the dump. + """ + db.session.commit() + for sibling in siblings: + if not _holds_statistics_dump(sibling): + continue + current = replace(context, unit=sibling) + if sibling.object_key is None: + return _block(current, RetentionBlockReason.ARCHIVE_OBJECT_MISSING, "no object key is recorded") + if context.archive is None: + return _failed(current, ARCHIVE_NOT_CONFIGURED) + try: + size = context.archive.object_size(sibling.object_key) + except KudosArchiveError as err: + return _failed(current, f"archive unavailable before drop: {err}") + if size is None: + return _block(current, RetentionBlockReason.ARCHIVE_OBJECT_MISSING, sibling.object_key) + return None + + +def _check_archive_before_drop(context: _StepContext, siblings: Sequence[CatalogUnit]) -> RetentionTickResult | None: + """Return the outcome that stops the drop when an archive object of the partition's units fails its check, else None.""" + currency_problem = _check_currency_archive(context, siblings) + if currency_problem is not None: + return currency_problem + return _check_statistics_dumps(context, siblings) + + def _drop_partition(context: _StepContext) -> RetentionTickResult: """Detach and drop the unit's partition and mark all of its units ``DROPPED``. @@ -1664,7 +1746,7 @@ def _drop_partition(context: _StepContext) -> RetentionTickResult: siblings = [sibling for sibling in load_catalog(context.unit.dataset) if sibling.partition_name == context.unit.partition_name] if not _partition_drop_ready(siblings): return _failed(context, "partition has units whose replacement is not committed") - archive_problem = _check_currency_archive(context, siblings) + archive_problem = _check_archive_before_drop(context, siblings) if archive_problem is not None: return archive_problem partition_exists = _relation_exists(context.unit.partition_name) @@ -1709,7 +1791,7 @@ def _drop_partition(context: _StepContext) -> RetentionTickResult: return _block(replace(context, unit=detached_late_unit[0]), RetentionBlockReason.LATE_ROWS, detail) if not keep_detached: # The recount can take a while, so the archive is checked again immediately before the removal. - archive_problem = _check_currency_archive(context, siblings) + archive_problem = _check_archive_before_drop(context, siblings) if archive_problem is not None: return archive_problem autocommit_connection.execute(text(f"DROP TABLE IF EXISTS {context.partition_sql}")) @@ -1842,8 +1924,20 @@ def _run_locked_tick( if not policy.enabled: return RetentionTickResult(outcome=RetentionTickOutcome.IDLE, detail="retention disabled") partitions = discover_partitions() + restored = restored_partitions(partitions, load_catalog()) + refusals: list[RetentionTickResult] = [] candidates: list[CatalogUnit] = [] for dataset in datasets: + restored_names = [entry["partition_name"] for entry in restored if entry["dataset"] == dataset] + if restored_names: + refusals.append( + RetentionTickResult( + outcome=RetentionTickOutcome.BLOCKED, + dataset=dataset, + detail=f"{RESTORED_PARTITION_REFUSAL}: {', '.join(restored_names)}", + ), + ) + continue unit = _next_unit( dataset, partitions, @@ -1860,6 +1954,8 @@ def _run_locked_tick( logger.info(f"Kudos retention holds {held_partitions} partitions") db.session.commit() if not candidates: + if refusals: + return refusals[0] return RetentionTickResult(outcome=RetentionTickOutcome.IDLE, detail="no actionable unit") failures: list[RetentionTickResult] = [] # A failed or deferred step makes no transition, so the other dataset's unit still gets this tick. An unreachable @@ -1869,6 +1965,8 @@ def _run_locked_tick( if result.outcome not in (RetentionTickOutcome.FAILED, RetentionTickOutcome.DEFERRED): return result failures.append(result) + if refusals: + return refusals[0] return failures[0] @@ -1964,12 +2062,16 @@ def _currency_unit_awaiting_drop(unit: CatalogUnit) -> bool: def _reconcile_archive_catalog(archive: KudosArchive) -> dict[ArchiveTier, int] | None: - """Clear ``object_key`` on units whose object is not in the archive, commit, and return each tier's stored bytes. + """Update the units whose object is not in the archive, commit, and return each tier's stored bytes. - A key whose object is gone points at nothing, whatever removed the object: an operator eviction, one that stopped - partway, or a delete outside the horde. A currency unit in ``EXPORTED``, ``VERIFIED`` or ``DETACHED`` also becomes - ``BLOCKED`` with ``ARCHIVE_OBJECT_MISSING``, because its rows are still online and the drop would otherwise remove - them without a copy. A listing failure skips the reconciliation for this tick. + ``archive evict`` clears the keys of the units it evicts, so a listed key without its object was removed outside + the horde, or by an eviction that stopped before it updated the catalog. A currency unit in ``EXPORTED``, + ``VERIFIED`` or ``DETACHED`` becomes ``BLOCKED`` with ``ARCHIVE_OBJECT_MISSING``, because its rows are still online + and the drop would otherwise remove them without a copy. A ``DUMPED`` statistics unit whose partition is attached + becomes ``DUMP_FAILED``, so the next ticks dump its rows again within the dump retry window. A detached one + becomes ``BLOCKED`` with ``ARCHIVE_OBJECT_MISSING``. A ``DROPPED`` statistics unit keeps its key, so the archive + audit keeps reporting the lost object. Every other unit loses the key. A listing failure skips the reconciliation + for this tick. Returns: Each tier's stored bytes, or None when a listing failed. @@ -1981,8 +2083,32 @@ def _reconcile_archive_catalog(archive: KudosArchive) -> dict[ArchiveTier, int] return None listed_keys = {key for sizes in sizes_by_tier.values() for key in sizes} for missing_unit in units_without_archive_objects(listed_keys): + if missing_unit.dataset == KudosRetentionDataset.STATISTICS and missing_unit.state == KudosRetentionState.DROPPED: + continue values: dict[str, object] = {"object_key": None, "updated": _utcnow()} - if _currency_unit_awaiting_drop(missing_unit): + if _holds_statistics_dump(missing_unit) and missing_unit.state == KudosRetentionState.DUMPED: + # The raw rows are still online, so the dump is repeated on the next tick, without the spacing that + # follows a failed attempt. The retry window starts now. + values |= { + "state": KudosRetentionState.DUMP_FAILED, + "error": f"{RetentionBlockReason.ARCHIVE_OBJECT_MISSING}: {missing_unit.object_key}", + "first_failed_at": missing_unit.first_failed_at or _utcnow(), + "last_dump_attempt_at": None, + } + # LOG CONTRACT: the line starts with RETENTION_DUMP_FAILED_LOG_PREFIX ("Kudos retention dump failed"). + logger.warning( + f"{RETENTION_DUMP_FAILED_LOG_PREFIX} for {missing_unit.source_range}: the archive does not hold " + f"{missing_unit.object_key}, so the unit is dumped again", + ) + elif _holds_statistics_dump(missing_unit): + values |= {"state": KudosRetentionState.BLOCKED, "error": str(RetentionBlockReason.ARCHIVE_OBJECT_MISSING)} + # LOG CONTRACT: the line starts with RETENTION_BLOCKED_LOG_PREFIX ("Kudos retention blocked"). + logger.warning( + f"{RETENTION_BLOCKED_LOG_PREFIX} {missing_unit.dataset} unit {missing_unit.source_range}: " + f"{RetentionBlockReason.ARCHIVE_OBJECT_MISSING}: the archive does not hold {missing_unit.object_key} " + f"(state {missing_unit.state})", + ) + elif _currency_unit_awaiting_drop(missing_unit): # The drop would remove the only copy of these postings, so the unit stops until it is exported again. values |= {"state": KudosRetentionState.BLOCKED, "error": str(RetentionBlockReason.ARCHIVE_OBJECT_MISSING)} # LOG CONTRACT: the line starts with RETENTION_BLOCKED_LOG_PREFIX ("Kudos retention blocked"). @@ -2210,9 +2336,10 @@ def recheckpoint_from_unit(unit_id: int) -> list[CatalogUnit]: def waive_unit_dump(unit_id: int) -> CatalogUnit: """Accept dropping a statistics unit without its archive dump, moving it to ``DUMP_WAIVED``, and commit. - Only a unit whose dump did not happen qualifies: ``DUMP_FAILED``, or ``BLOCKED`` with a reason in - ``WAIVABLE_DUMP_BLOCK_REASONS`` (retries exhausted, or the statistics budget). Its rollup is already committed, so - only the raw rows are lost. The ``error`` text is kept as the record of why. + Only a unit without a dump qualifies: ``DUMP_FAILED``, or ``BLOCKED`` with a reason in + ``WAIVABLE_DUMP_BLOCK_REASONS`` (retries exhausted, the statistics budget, or a dump missing from the archive). Its + rollup is already committed, so only the raw rows are lost. The ``error`` text is kept as the record of why. A unit + of a detached partition is refused until the partition is reattached, so the drop starts from an attached partition. Args: unit_id: The catalog id of the unit. @@ -2221,7 +2348,8 @@ def waive_unit_dump(unit_id: int) -> CatalogUnit: The unit as waived. Raises: - ValueError: No unit has that id, it is not a statistics unit, or it is neither failed nor blocked for a waivable reason. + ValueError: No unit has that id, it is not a statistics unit, it is neither failed nor blocked for a waivable + reason, or its partition is detached. RetentionLockBusyError: A tick holds the retention lock. """ with operator_retention_lock(): @@ -2230,6 +2358,8 @@ def waive_unit_dump(unit_id: int) -> CatalogUnit: raise ValueError(f"No kudos retention unit {unit_id}") if row.dataset != KudosRetentionDataset.STATISTICS: raise ValueError(f"Kudos retention unit {unit_id} is a {row.dataset} unit, and only statistics dumps can be waived") + if row.detached_at is not None: + raise ValueError(f"Kudos retention unit {unit_id} belongs to detached partition {row.partition_name}. Reattach it first") waivable = row.state == KudosRetentionState.BLOCKED and row.error in WAIVABLE_DUMP_BLOCK_REASONS if row.state != KudosRetentionState.DUMP_FAILED and not waivable: raise ValueError( @@ -2243,6 +2373,64 @@ def waive_unit_dump(unit_id: int) -> CatalogUnit: return _catalog_unit(row) +@dataclass(frozen=True, kw_only=True) +class StatisticsEviction: + """Represents the outcome of ``evict_statistics_dumps``.""" + + deleted_keys: list[str] + """The deleted object keys, oldest first.""" + tier_bytes: int + """The stored bytes of the statistics tier after the eviction.""" + + +def evict_statistics_dumps(archive: KudosArchive, *, budget_bytes: int) -> StatisticsEviction: + """Delete the oldest statistics dumps of dropped units until the tier fits ``budget_bytes``, and clear their keys. + + The dump of a unit that is not ``DROPPED`` is kept, because its raw rows are still online and its drop relies on + the dump. Objects that are not recorded by any unit are deleted like dropped dumps. The keys of the evicted units are cleared in + the catalog, so the tick and the archive audit treat any other missing dump as lost. The eviction runs under the + retention lock, because an export measures its allowance from the tier size. + + Args: + archive: The archive client. + budget_bytes: The byte budget of the statistics tier. + + Returns: + The deleted keys and the stored bytes of the tier afterwards, which stay above the budget when the kept dumps + alone exceed it. + + Raises: + RetentionLockBusyError: A tick holds the retention lock. + KudosArchiveError: The listing failed, before anything was deleted. + ArchiveEvictionError: A delete failed. The keys deleted before it are cleared in the catalog first. + """ + with operator_retention_lock(): + kept_keys = { + catalog_unit.object_key + for catalog_unit in load_catalog(KudosRetentionDataset.STATISTICS) + if catalog_unit.object_key is not None and catalog_unit.state != KudosRetentionState.DROPPED + } + try: + deleted_keys = archive.enforce_budget(budget_bytes=budget_bytes, kept_keys=kept_keys) + except ArchiveEvictionError as err: + _clear_evicted_keys(err.deleted_keys) + raise + _clear_evicted_keys(deleted_keys) + return StatisticsEviction(deleted_keys=deleted_keys, tier_bytes=archive.tier_bytes(ArchiveTier.STATISTICS)) + + +def _clear_evicted_keys(deleted_keys: Sequence[str]) -> None: + if not deleted_keys: + return + db.session.execute( + update(KudosRetentionUnit) + .where(KudosRetentionUnit.dataset == KudosRetentionDataset.STATISTICS, KudosRetentionUnit.object_key.in_(deleted_keys)) + .values(object_key=None, updated=_utcnow()), + ) + db.session.commit() + logger.warning(f"Kudos retention cleared the object keys of {len(deleted_keys)} evicted statistics dumps") + + def _attached_partition_names() -> set[str]: return {partition.name for partition in discover_partitions()} @@ -2490,7 +2678,7 @@ def audit_archive( def retention_outcome_warning(result: RetentionTickResult) -> str | None: - """Return the warning to log for a tick that blocked a unit or failed a statistics dump, or None otherwise. + """Return the warning to log for a tick that blocked a unit or a dataset or failed a statistics dump, or None otherwise. The lines start with ``RETENTION_BLOCKED_LOG_PREFIX`` or ``RETENTION_DUMP_FAILED_LOG_PREFIX``. A unit waived by the dump step because no archive is configured gets the dump-failed line, once, on that transition. @@ -2504,6 +2692,8 @@ def retention_outcome_warning(result: RetentionTickResult) -> str | None: upper=result.unit.upper, ) ) + if result.outcome == RetentionTickOutcome.BLOCKED and result.unit is None: + return f"{RETENTION_BLOCKED_LOG_PREFIX} {result.dataset} dataset: {result.detail}" if result.outcome == RetentionTickOutcome.BLOCKED: return f"{RETENTION_BLOCKED_LOG_PREFIX} {result.dataset} unit {source_range}: {result.detail}" if result.to_state in (KudosRetentionState.DUMP_FAILED, KudosRetentionState.DUMP_WAIVED): @@ -2555,7 +2745,7 @@ class AccountingRetentionDisclosure: def retention_health(*, now: datetime | None = None, policy: KudosRetentionPolicy = KUDOS_RETENTION_POLICY) -> KudosRetentionHealth: - """Return the retention job's progress: eligible backlog age, unit counts, last step age, coverage and held drops. + """Return the retention job's progress: backlog age, unit counts, last step age, coverage, held, detached and restored partitions. Args: now: The reference time (naive UTC); the current time when None. @@ -2603,6 +2793,7 @@ def retention_health(*, now: datetime | None = None, policy: KudosRetentionPolic held_partitions=drop_ready_partition_count(catalog) if policy.hold_drops else 0, drop_grace_days=policy.drop_grace / timedelta(days=1), detached_partitions=detached_partitions(catalog, drop_grace=policy.drop_grace), + restored_partitions=restored_partitions(partitions, catalog), ) @@ -2629,6 +2820,8 @@ def _partition_contains(partition: PartitionBounds, moment: datetime) -> bool: __all__ = [ + "StatisticsEviction", + "evict_statistics_dumps", "ARCHIVE_BUDGET_ENV", "ARCHIVE_STATISTICS_BUDGET_ENV", "ARCHIVE_TIERS", @@ -2650,6 +2843,9 @@ __all__ = [ "DUMP_RETRY_SPACING", "DUMP_RETRY_WINDOW_ENV", "FUTURE_REFERENCE_REFUSAL", + "RESTORED_PARTITION_REFUSAL", + "RestoredPartitionHealth", + "restored_partitions", "HOLD_DROPS_ENV", "KUDOS_RETENTION_POLICY", "PARTITION_WIDTH_ENV", diff --git a/horde/database/threads.py b/horde/database/threads.py index 24621b46..c4886879 100644 --- a/horde/database/threads.py +++ b/horde/database/threads.py @@ -916,6 +916,7 @@ def kudos_retention_tick() -> None: kudos_retention_held_partitions, kudos_retention_oldest_eligible_age, kudos_retention_partition_coverage, + kudos_retention_restored_partitions, kudos_retention_steps, ) @@ -960,6 +961,7 @@ def kudos_retention_tick() -> None: kudos_retention_blocked_units.set(blocked_count, {"horde.kudos.block_reason": str(block_reason)}) kudos_retention_partition_coverage.set(1 if health["partition_coverage_ok"] else 0) kudos_retention_held_partitions.set(health["held_partitions"]) + kudos_retention_restored_partitions.set(len(health["restored_partitions"])) if archive is None: return for tier in ArchiveTier: diff --git a/horde/kudos_archive.py b/horde/kudos_archive.py index e29609a3..20aa589d 100644 --- a/horde/kudos_archive.py +++ b/horde/kudos_archive.py @@ -31,7 +31,7 @@ import json import os import re import zlib -from collections.abc import Iterator, Mapping +from collections.abc import Collection, Iterator, Mapping from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from decimal import Decimal @@ -1069,15 +1069,16 @@ class KudosArchive: return None return ArchiveVersionUsage(noncurrent_bytes=noncurrent_bytes, delete_markers=delete_markers) - def enforce_budget(self, *, budget_bytes: int) -> list[str]: + def enforce_budget(self, *, budget_bytes: int, kept_keys: Collection[str] = frozenset()) -> list[str]: """Delete the oldest statistic objects, by unit upper bound, until the statistics tier fits ``budget_bytes``. An operator tool. The retention tick does not call it, and currency objects are never deleted. Each deleted object's manifest is deleted with it. Deletes go out one batch at a time, so a failure leaves the earlier - batches deleted and reports them. + batches deleted and reports them. The tier can stay above the budget when ``kept_keys`` holds too much of it. Args: budget_bytes: The byte budget of the statistics tier. + kept_keys: Object keys that are not deleted, such as the dumps of units whose raw rows are still online. Returns: The deleted object keys, oldest first. @@ -1092,6 +1093,8 @@ class KudosArchive: for listed in sorted(statistics_objects, key=lambda candidate: (unit_upper_from_key(candidate.key), candidate.key)): if total <= budget_bytes: break + if listed.key in kept_keys: + continue candidates.append(listed.key) total -= listed.size evicted: list[str] = [] diff --git a/horde/metrics.py b/horde/metrics.py index 16f463a6..9428f02f 100644 --- a/horde/metrics.py +++ b/horde/metrics.py @@ -1035,6 +1035,11 @@ kudos_retention_held_partitions = logfire.metric_gauge( unit="{partition}", description="Drop-ready kudos event partitions kept attached because KUDOS_RETENTION_HOLD_DROPS is set", ) +kudos_retention_restored_partitions = logfire.metric_gauge( + "horde.kudos.retention.restored_partitions", + unit="{partition}", + description="Attached kudos event partitions whose catalog units are dropped. Their dataset does not make a retention transition", +) kudos_retention_dump_failures = logfire.metric_counter( "horde.kudos.retention.dump_failures", unit="{failure}", diff --git a/sql_statements/5.1.16.txt b/sql_statements/5.1.16.txt index 92e435db..54aed47f 100644 --- a/sql_statements/5.1.16.txt +++ b/sql_statements/5.1.16.txt @@ -613,10 +613,26 @@ SELECT pg_temp.kudos_validate_legacy_range('kudos_ledger'); START TRANSACTION; SET LOCAL lock_timeout = '5s'; --- The application locks a users row and then inserts into kudos_ledger, whose foreign key check locks that users --- row again. The swap takes ACCESS EXCLUSIVE on kudos_ledger at the rename and SHARE ROW EXCLUSIVE on users when it --- adds the foreign key to the new parent. Taking the users lock first follows the application's order, so an --- application transaction waits for the swap or the swap waits for it, and the two cannot deadlock. +-- The swap takes ACCESS EXCLUSIVE on kudos_ledger at the rename and SHARE ROW EXCLUSIVE on users when it adds the +-- foreign key to the new parent. The kudos applier claims kudos_ledger rows and then updates users, the reverse +-- order. The swap therefore first takes the applier's transaction advisory lock (KUDOS_APPLIER_LOCK in +-- horde/database/kudos_db.py). The applier only tries that lock, so it skips its cycles until the swap commits. +SELECT pg_advisory_xact_lock(1263879247); +-- A request transaction that inserts into kudos_ledger before it updates users can still form a lock cycle with the +-- swap. The swap takes its kudos_ledger lock right after the users lock, and each lock wait of the swap gives up +-- before deadlock_timeout. In such a cycle the swap fails with lock_not_available before PostgreSQL runs the deadlock +-- check in the waiting request, so the request continues and the file is rerun. +DO $$ +BEGIN + IF current_setting('deadlock_timeout')::interval <= interval '500 milliseconds' THEN + RAISE EXCEPTION 'deadlock_timeout is %, and the kudos_ledger swap needs it above its 500ms lock_timeout', + current_setting('deadlock_timeout'); + END IF; +END; +$$; +SET LOCAL lock_timeout = '500ms'; +-- Taking the users lock before kudos_ledger follows the order of a request that locks a users row and then inserts +-- into kudos_ledger, whose foreign key check locks that users row again. LOCK TABLE users IN SHARE ROW EXCLUSIVE MODE; SELECT pg_temp.kudos_normalize_columns('kudos_ledger'); SELECT pg_temp.kudos_swap_normalized_index('kudos_ledger', 'ix_kudos_ledger_unapplied'); diff --git a/tests/integration/test_kudos_retention_migration.py b/tests/integration/test_kudos_retention_migration.py index 638682e7..f2d1cfb3 100644 --- a/tests/integration/test_kudos_retention_migration.py +++ b/tests/integration/test_kudos_retention_migration.py @@ -9,7 +9,7 @@ from __future__ import annotations import re import threading import time as clock -from collections.abc import Iterator, Mapping +from collections.abc import Callable, Iterator, Mapping from datetime import time, timedelta from pathlib import Path from typing import Any @@ -229,11 +229,17 @@ def _schema_engine(pg_dsn: str, schema_name: str) -> sqlalchemy.Engine: ) -def _apply_migration(connection: sqlalchemy.Connection, *, variables: Mapping[str, str] | None = None) -> list[str]: +def _apply_migration( + connection: sqlalchemy.Connection, + *, + variables: Mapping[str, str] | None = None, + before_statement: Callable[[str], None] | None = None, +) -> list[str]: """Run the migration as psql does, including its ``\\gexec`` statements, and return its ``\\echo`` lines. psql sends the query that ends at a ``\\gexec`` line and then runs every value it returns as a statement of its own, outside any transaction block. ``variables`` are the ``-v name=value`` options of the psql command line. + ``before_statement`` is called with each statement before it is sent, so a test can act at a point of the file. """ script, echoed_lines = _expand_psql_variables(MIGRATION_PATH.read_text(), dict(variables or {})) segments = GEXEC_LINE.split(script) @@ -241,11 +247,15 @@ def _apply_migration(connection: sqlalchemy.Connection, *, variables: Mapping[st statements = [statement for statement in sqlparse.split(segment) if sqlparse.format(statement, strip_comments=True).strip()] generating_query = statements.pop() if segment_index < len(segments) - 1 else None for statement in statements: + if before_statement is not None: + before_statement(statement) connection.execute(sqlalchemy.text(statement)) if generating_query is None: continue for generated_row in connection.execute(sqlalchemy.text(generating_query)).all(): for generated_statement in generated_row: + if before_statement is not None: + before_statement(generated_statement) connection.execute(sqlalchemy.text(generated_statement)) return echoed_lines @@ -630,6 +640,150 @@ def test_rerun_refuses_a_legacy_range_it_cannot_read(unconverted_schema: sqlalch assert _scalar(connection, "SELECT relkind FROM pg_class WHERE oid = CAST('kudos_ledger' AS regclass)") == "r" +LEDGER_SWAP_START = "START TRANSACTION;" +"""The statement that opens the ``kudos_ledger`` swap, the first transaction block of the file.""" +UNAPPLIED_POSTING = ( + "INSERT INTO kudos_ledger (event_id, entry_type, user_id, amount) VALUES ('00000000-0000-0000-0000-0000000000e1', 'AWARD', 1, 4)" +) +"""A posting as a request writes it, unapplied and created now.""" + + +def _second_session(pg_dsn: str, connection: sqlalchemy.Connection) -> tuple[sqlalchemy.Engine, sqlalchemy.Connection]: + """Return an engine and a connection to the schema of ``connection``, for a second session in the test.""" + schema_name = _scalar(connection, "SELECT current_schema()") + engine = _schema_engine(pg_dsn, schema_name) + return engine, engine.connect() + + +def _wait_for_lock_wait(observer: sqlalchemy.Connection, backend_pid: int, *, timeout_seconds: float = 5) -> None: + """Return once the backend ``backend_pid`` waits for a lock, or fail the test after ``timeout_seconds``.""" + deadline = clock.monotonic() + timeout_seconds + while clock.monotonic() < deadline: + if _scalar(observer, "SELECT count(*) FROM pg_locks WHERE pid = :pid AND NOT granted", pid=backend_pid): + return + clock.sleep(0.01) + pytest.fail(f"backend {backend_pid} did not wait for a lock within {timeout_seconds} seconds") + + +def _run_migration_in_thread( + connection: sqlalchemy.Connection, + before_statement: Callable[[str], None], +) -> tuple[threading.Thread, list[BaseException]]: + """Start the migration on ``connection`` in a thread and return the thread and the list its error is added to.""" + errors: list[BaseException] = [] + + def run() -> None: + try: + _apply_migration(connection, before_statement=before_statement) + except BaseException as err: # noqa: BLE001 - asserted in the test thread + errors.append(err) + + thread = threading.Thread(target=run) + thread.start() + return thread, errors + + +def _wait_for_flag(flag: list[bool], thread: threading.Thread, errors: list[BaseException]) -> None: + """Return once ``flag`` has an entry, or fail the test when ``thread`` ended first or 30 seconds passed.""" + deadline = clock.monotonic() + 30 + while not flag and clock.monotonic() < deadline and thread.is_alive(): + clock.sleep(0.01) + assert flag, errors + + +def test_ledger_swap_waits_for_the_applier_lock_instead_of_deadlocking(pg_dsn: str, unconverted_schema: sqlalchemy.Connection) -> None: + """An applier cycle that claims ledger rows and then updates users finishes, and the swap commits after it. + + The applier holds its advisory lock and a row lock on ``kudos_ledger`` when the swap starts. The swap waits for the + advisory lock before it locks ``users``, so the update of ``users`` by the applier does not wait for the swap. + """ + from horde.database.kudos_db import KUDOS_APPLIER_LOCK + + connection = unconverted_schema + connection.execute(sqlalchemy.text("INSERT INTO users (id) VALUES (1)")) + connection.execute(sqlalchemy.text(UNAPPLIED_POSTING)) + migration_pid = _scalar(connection, "SELECT pg_backend_pid()") + applier_engine, applier = _second_session(pg_dsn, connection) + applier_lock_taken: list[bool] = [] + + def start_the_applier_cycle(statement: str) -> None: + if statement.strip() != LEDGER_SWAP_START or applier_lock_taken: + return + applier.execute(sqlalchemy.text("BEGIN")) + applier_lock_taken.append(_scalar(applier, "SELECT pg_try_advisory_xact_lock(:key)", key=KUDOS_APPLIER_LOCK)) + applier.execute(sqlalchemy.text("SELECT id FROM kudos_ledger WHERE NOT applied FOR UPDATE SKIP LOCKED")).all() + + try: + migration_thread, migration_errors = _run_migration_in_thread(connection, start_the_applier_cycle) + try: + _wait_for_flag(applier_lock_taken, migration_thread, migration_errors) + assert applier_lock_taken == [True] + _wait_for_lock_wait(applier, migration_pid) + applier.execute(sqlalchemy.text("UPDATE users SET id = id WHERE id = 1")) + applier.execute(sqlalchemy.text("COMMIT")) + finally: + migration_thread.join(timeout=30) + finally: + applier.close() + applier_engine.dispose() + assert not migration_thread.is_alive() + assert not migration_errors, migration_errors + assert _scalar(connection, "SELECT relkind FROM pg_class WHERE oid = CAST('kudos_ledger' AS regclass)") == "p" + + +def test_ledger_swap_gives_up_before_a_request_deadlocks_with_it(pg_dsn: str, unconverted_schema: sqlalchemy.Connection) -> None: + """A request that inserted a posting and then updates users completes, and the swap fails and is rerun. + + The swap holds the ``users`` lock and waits for the lock of the request on ``kudos_ledger`` when the request + updates ``users``. The lock wait of the swap ends before ``deadlock_timeout``, so the swap fails with + ``lock_not_available`` and the request is not chosen as the deadlock victim. + """ + connection = unconverted_schema + connection.execute(sqlalchemy.text("INSERT INTO users (id) VALUES (1)")) + migration_pid = _scalar(connection, "SELECT pg_backend_pid()") + request_engine, request = _second_session(pg_dsn, connection) + request_started: list[bool] = [] + + def start_the_request(statement: str) -> None: + if statement.strip() != LEDGER_SWAP_START or request_started: + return + request.execute(sqlalchemy.text("BEGIN")) + request.execute(sqlalchemy.text(UNAPPLIED_POSTING)) + request_started.append(True) + + try: + migration_thread, migration_errors = _run_migration_in_thread(connection, start_the_request) + try: + _wait_for_flag(request_started, migration_thread, migration_errors) + _wait_for_lock_wait(request, migration_pid) + request.execute(sqlalchemy.text("UPDATE users SET id = id WHERE id = 1")) + request.execute(sqlalchemy.text("COMMIT")) + finally: + migration_thread.join(timeout=30) + finally: + request.close() + request_engine.dispose() + assert not migration_thread.is_alive() + assert len(migration_errors) == 1 + assert isinstance(migration_errors[0], sqlalchemy.exc.OperationalError) + assert getattr(migration_errors[0].orig, "pgcode", None) == "55P03" + connection.execute(sqlalchemy.text("ROLLBACK")) + assert _scalar(connection, "SELECT relkind FROM pg_class WHERE oid = CAST('kudos_ledger' AS regclass)") == "r" + _apply_migration(connection) + assert _scalar(connection, "SELECT relkind FROM pg_class WHERE oid = CAST('kudos_ledger' AS regclass)") == "p" + assert _scalar(connection, "SELECT count(*) FROM kudos_ledger") == 1 + + +def test_ledger_swap_refuses_a_deadlock_timeout_at_or_below_its_lock_timeout(unconverted_schema: sqlalchemy.Connection) -> None: + """The lock waits of the swap must end before the deadlock check of a waiting request, so a short deadlock_timeout stops the run.""" + connection = unconverted_schema + connection.execute(sqlalchemy.text("SET deadlock_timeout = '500ms'")) + with pytest.raises(sqlalchemy.exc.InternalError, match="deadlock_timeout is 500ms"): + _apply_migration(connection) + connection.execute(sqlalchemy.text("ROLLBACK")) + assert _scalar(connection, "SELECT relkind FROM pg_class WHERE oid = CAST('kudos_ledger' AS regclass)") == "r" + + def _catalog(connection: sqlalchemy.Connection, schema_name: str) -> dict[str, Any]: """Return every column attribute, constraint, index and partition key of the compared tables. diff --git a/tests/unit/test_kudos_retention.py b/tests/unit/test_kudos_retention.py index ec6bd78c..10a7f178 100644 --- a/tests/unit/test_kudos_retention.py +++ b/tests/unit/test_kudos_retention.py @@ -1901,6 +1901,43 @@ def test_verified_currency_unit_whose_object_vanishes_blocks_and_is_exported_aga assert not _relation_present(db_session, partition) +def test_partition_restored_under_its_dropped_name_stops_its_dataset_until_it_is_detached( + db_session: scoped_session[Session], + make_partition: Callable[..., str], + make_user: MakeUser, +) -> None: + """A dropped currency partition restored and attached under the same name stops currency, and statistics proceed.""" + from horde.database.kudos_retention import RESTORED_PARTITION_REFUSAL, retention_health, retention_outcome_warning, run_retention_tick + + partition, _, archive = _verified_currency_week(db_session, make_partition, make_user) + assert _run(archive).to_state == KudosRetentionState.DROPPED + assert not _relation_present(db_session, partition) + restored = make_partition("kudos_ledger", MONDAY, MONDAY + WEEK) + assert restored == partition + statistics_partition = make_partition("kudos_stat_events", MONDAY, MONDAY + WEEK) + db_session.add(_stat_event(MONDAY + timedelta(hours=1), "2")) + db_session.commit() + + stepped = _run(archive) + assert (stepped.dataset, stepped.unit.partition_name, stepped.to_state) == ( + STATISTICS, + statistics_partition, + KudosRetentionState.ROLLED_UP, + ) + refused = run_retention_tick(now=AFTER_WINDOW, policy=_policy(), archive=archive, dataset=CURRENCY) + assert (refused.outcome, refused.dataset, refused.unit) == (RetentionTickOutcome.BLOCKED, CURRENCY, None) + assert refused.detail == f"{RESTORED_PARTITION_REFUSAL}: {partition}" + assert retention_outcome_warning(refused) == f"Kudos retention blocked currency dataset: {RESTORED_PARTITION_REFUSAL}: {partition}" + assert retention_health(now=AFTER_WINDOW)["restored_partitions"] == [{"dataset": CURRENCY, "partition_name": partition}] + assert _attached(db_session, partition) + assert _unit_of(db_session, partition).state == KudosRetentionState.DROPPED + + db_session.execute(text(f"ALTER TABLE kudos_ledger DETACH PARTITION {partition}")) + db_session.commit() + assert retention_health(now=AFTER_WINDOW)["restored_partitions"] == [] + assert run_retention_tick(now=AFTER_WINDOW, policy=_policy(), archive=archive, dataset=CURRENCY).outcome == RetentionTickOutcome.IDLE + + def test_drop_blocks_a_currency_unit_without_an_object_key( db_session: scoped_session[Session], make_partition: Callable[..., str], @@ -1916,28 +1953,151 @@ def test_drop_blocks_a_currency_unit_without_an_object_key( assert _attached(db_session, partition) -def test_tick_clears_object_keys_whose_objects_are_gone_and_reports_tier_bytes( - db_session: scoped_session[Session], - make_partition: Callable[..., str], -) -> None: +def _dumped_statistics_week(db_session: scoped_session[Session], make_partition: Callable[..., str]) -> tuple[str, Any, Any]: + """Create a statistics week with one event and step it to ``DUMPED`` with drops held. Returns the partition, store + and archive.""" partition = make_partition("kudos_stat_events", MONDAY, MONDAY + WEEK) db_session.add(_stat_event(MONDAY + timedelta(hours=1), "2")) db_session.commit() store = _FakeObjectStore() archive = _fake_archive(store) - assert _run(archive).to_state == KudosRetentionState.ROLLED_UP - dumped = _run(archive, hold_drops=True) - assert dumped.to_state == KudosRetentionState.DUMPED + states = [_run(archive, hold_drops=True).to_state for _ in range(2)] + assert states == [KudosRetentionState.ROLLED_UP, KudosRetentionState.DUMPED] + return partition, store, archive + + +def _delete_object(store: Any, key: str) -> None: + from horde.kudos_archive import manifest_key + + del store.objects[key] + del store.objects[manifest_key(key)] + + +def test_tick_dumps_again_a_statistics_unit_whose_dump_vanished_and_reports_tier_bytes( + db_session: scoped_session[Session], + make_partition: Callable[..., str], +) -> None: + """A dump removed outside the horde while the raw rows are online is repeated, and the drop then proceeds.""" + partition, store, archive = _dumped_statistics_week(db_session, make_partition) key = _unit_of(db_session, partition).object_key assert key is not None - assert dumped.archive_tier_bytes == {ArchiveTier.CURRENCY: 0, ArchiveTier.STATISTICS: len(store.objects[key])} - # An eviction outside the tick removes the object, and the next tick clears the key that points at nothing. - del store.objects[key] + _delete_object(store, key) idle = _run(archive, hold_drops=True) assert idle.archive_tier_bytes == {ArchiveTier.CURRENCY: 0, ArchiveTier.STATISTICS: 0} unit = _unit_of(db_session, partition) - assert (unit.state, unit.object_key) == (KudosRetentionState.DUMPED, None) + assert (unit.state, unit.object_key) == (KudosRetentionState.DUMP_FAILED, None) + assert unit.error == f"{RetentionBlockReason.ARCHIVE_OBJECT_MISSING}: {key}" + assert unit.first_failed_at is not None + + redumped = _run(archive, hold_drops=True) + assert redumped.to_state == KudosRetentionState.DUMPED + assert _unit_of(db_session, partition).object_key == key + assert redumped.archive_tier_bytes is not None and redumped.archive_tier_bytes[ArchiveTier.STATISTICS] > 0 + assert _run(archive).to_state == KudosRetentionState.DROPPED + + +@pytest.mark.parametrize("resolution", ["unblock", "waive"]) +def test_drop_blocks_a_statistics_unit_whose_dump_vanished_since_the_last_listing( + db_session: scoped_session[Session], + make_partition: Callable[..., str], + resolution: str, +) -> None: + """The drop checks each dump, so the raw rows are kept when a dump is removed after the last tick.""" + from horde.database.kudos_retention import unblock_unit, waive_unit_dump + + partition, store, archive = _dumped_statistics_week(db_session, make_partition) + _delete_object(store, _unit_of(db_session, partition).object_key) + blocked = _run(archive) + assert blocked.outcome == RetentionTickOutcome.BLOCKED + assert blocked.detail.startswith(str(RetentionBlockReason.ARCHIVE_OBJECT_MISSING)) + assert _attached(db_session, partition) + + unit_id = _unit_of(db_session, partition).id + if resolution == "unblock": + assert unblock_unit(unit_id).state == KudosRetentionState.PENDING + states = [_run(archive).to_state for _ in range(3)] + assert states == [KudosRetentionState.ROLLED_UP, KudosRetentionState.DUMPED, KudosRetentionState.DROPPED] + else: + assert waive_unit_dump(unit_id).state == KudosRetentionState.DUMP_WAIVED + assert _run(archive).to_state == KudosRetentionState.DROPPED + assert not _relation_present(db_session, partition) + + +def test_detached_statistics_unit_whose_dump_vanished_blocks_until_reattached( + db_session: scoped_session[Session], + make_partition: Callable[..., str], +) -> None: + """A dump lost during the drop grace blocks the unit, and the waiver is refused until the partition is reattached.""" + from horde.database.kudos_retention import reattach_partition, waive_unit_dump + + partition, store, archive = _dumped_statistics_week(db_session, make_partition) + assert _run(archive, drop_grace=timedelta(days=7)).to_state == KudosRetentionState.DETACHED + _delete_object(store, _unit_of(db_session, partition).object_key) + _run(archive, drop_grace=timedelta(days=7)) + unit = _unit_of(db_session, partition) + assert (unit.state, unit.error) == (KudosRetentionState.BLOCKED, RetentionBlockReason.ARCHIVE_OBJECT_MISSING) + assert unit.detached_at is not None + with pytest.raises(ValueError, match="Reattach it first"): + waive_unit_dump(unit.id) + # The grace has passed, but a blocked unit keeps the detached partition. + _run(archive, AFTER_WINDOW + timedelta(days=8), drop_grace=timedelta(days=7)) + assert _relation_present(db_session, partition) + + reattached = reattach_partition(unit.id) + assert [(reattached_unit.state, reattached_unit.error) for reattached_unit in reattached] == [ + (KudosRetentionState.BLOCKED, RetentionBlockReason.ARCHIVE_OBJECT_MISSING), + ] + assert _attached(db_session, partition) + assert waive_unit_dump(unit.id).state == KudosRetentionState.DUMP_WAIVED + + +def test_dropped_statistics_unit_keeps_the_key_of_a_vanished_dump_so_the_audit_reports_it( + db_session: scoped_session[Session], + make_partition: Callable[..., str], +) -> None: + from horde.database.kudos_retention import MissingArchiveObject, audit_archive + + partition, store, archive = _dumped_statistics_week(db_session, make_partition) + assert _run(archive).to_state == KudosRetentionState.DROPPED + key = _unit_of(db_session, partition).object_key + _delete_object(store, key) + for _ in range(2): + _run(archive) + assert _unit_of(db_session, partition).object_key == key + missing = [record for record in audit_archive(archive) if isinstance(record, MissingArchiveObject)] + assert [(record.object_key, record.state) for record in missing] == [(key, KudosRetentionState.DROPPED)] + + +def test_eviction_keeps_the_dumps_of_units_not_dropped_and_clears_the_keys_it_deletes( + db_session: scoped_session[Session], + make_partition: Callable[..., str], +) -> None: + """Evicted dumps are not reported as lost, and a dump whose raw rows are still online is never evicted.""" + from horde.database.kudos_retention import MissingArchiveObject, audit_archive, evict_statistics_dumps + + dropped_partition = make_partition("kudos_stat_events", MONDAY, MONDAY + WEEK) + held_partition = make_partition("kudos_stat_events", MONDAY + WEEK, MONDAY + 2 * WEEK) + db_session.add_all([_stat_event(MONDAY + timedelta(hours=1), "2"), _stat_event(MONDAY + WEEK + timedelta(hours=1), "3")]) + db_session.commit() + store = _FakeObjectStore() + archive = _fake_archive(store) + states = [_run(archive).to_state for _ in range(3)] + assert states == [KudosRetentionState.ROLLED_UP, KudosRetentionState.DUMPED, KudosRetentionState.DROPPED] + states = [_run(archive, hold_drops=True).to_state for _ in range(2)] + assert states == [KudosRetentionState.ROLLED_UP, KudosRetentionState.DUMPED] + dropped_key = _unit_of(db_session, dropped_partition).object_key + held_key = _unit_of(db_session, held_partition).object_key + + eviction = evict_statistics_dumps(archive, budget_bytes=0) + assert eviction.deleted_keys == [dropped_key] + assert eviction.tier_bytes == archive.tier_bytes(ArchiveTier.STATISTICS) > 0 + assert held_key in store.objects and dropped_key not in store.objects + assert _unit_of(db_session, dropped_partition).object_key is None + assert _unit_of(db_session, held_partition).object_key == held_key + assert [record for record in audit_archive(archive) if isinstance(record, MissingArchiveObject)] == [] + assert _run(archive, hold_drops=True).outcome == RetentionTickOutcome.IDLE + assert _unit_of(db_session, held_partition).state == KudosRetentionState.DUMPED def test_tick_aborts_multipart_uploads_older_than_a_day(db_session: scoped_session[Session]) -> None: @@ -2292,7 +2452,7 @@ def test_drop_grace_is_measured_with_the_rehearsal_reference_time( db_session: scoped_session[Session], current_week_partition: datetime, ) -> None: - from horde.database.kudos_retention import run_retention_tick + from horde.database.kudos_retention import RESTORED_PARTITION_REFUSAL, run_retention_tick rehearsal_partition = "kudos_stat_events_p_rehearsal" real_now = datetime.now(UTC).replace(tzinfo=None) @@ -2318,9 +2478,13 @@ def test_drop_grace_is_measured_with_the_rehearsal_reference_time( tick_until_idle(as_of + GRACE - timedelta(minutes=1)) assert rehearsal_state() == KudosRetentionState.DETACHED assert _relation_present(db_session, rehearsal_partition) - tick_until_idle(as_of + GRACE) + # The rehearsal also drops the current week that partition maintenance created again under its weekly name, + # and maintenance creates it once more. That table has dropped catalog units, so the dataset then blocks. + results = [run_retention_tick(now=as_of + GRACE, policy=policy, dataset=STATISTICS, confirm_live_data_loss=True) for _ in range(20)] assert rehearsal_state() == KudosRetentionState.DROPPED assert not _relation_present(db_session, rehearsal_partition) + assert results[-1].outcome == RetentionTickOutcome.BLOCKED + assert results[-1].detail.startswith(RESTORED_PARTITION_REFUSAL) finally: # Partitions the rehearsal detached are no longer attached, so the fixture's teardown does not see them. db_session.rollback() diff --git a/tools/kudos_ledger_admin.py b/tools/kudos_ledger_admin.py index c0270eea..59f93fbf 100644 --- a/tools/kudos_ledger_admin.py +++ b/tools/kudos_ledger_admin.py @@ -40,8 +40,8 @@ from horde.database.kudos_retention import ( # noqa: E402 RetentionLockBusyError, audit_archive, daily_statement, + evict_statistics_dumps, load_catalog, - operator_retention_lock, parse_rehearsal_as_of, reattach_partition, recheckpoint_from_unit, @@ -148,7 +148,7 @@ def _parser() -> argparse.ArgumentParser: ) evict = archive_commands.add_parser( "evict", - help="Delete the oldest statistics dumps until the tier fits a byte budget. Currency objects are never evicted", + help="Delete the oldest statistics dumps of dropped units until the tier fits a byte budget. Currency objects are never evicted", ) evict.add_argument("--tier", required=True, choices=[str(ArchiveTier.STATISTICS)]) evict.add_argument("--budget-bytes", dest="budget_bytes", type=int, required=True) @@ -179,15 +179,23 @@ def _archive_command(options: argparse.Namespace, archive_client: KudosArchive) return if options.archive_command == "evict": try: - # Eviction changes the tier size an export measures its allowance from, so it runs under the same lock. - with operator_retention_lock(): - deleted = archive_client.enforce_budget(budget_bytes=options.budget_bytes) + eviction = evict_statistics_dumps(archive_client, budget_bytes=options.budget_bytes) except RetentionLockBusyError as err: raise SystemExit(str(err)) from err except ArchiveEvictionError as err: print(json.dumps({"tier": options.tier, "budget_bytes": options.budget_bytes, "deleted": err.deleted_keys, "error": str(err)})) raise SystemExit(1) from err - print(json.dumps({"tier": options.tier, "budget_bytes": options.budget_bytes, "deleted": deleted})) + eviction_output: dict[str, object] = { + "tier": options.tier, + "budget_bytes": options.budget_bytes, + "deleted": eviction.deleted_keys, + "tier_bytes": eviction.tier_bytes, + } + if eviction.tier_bytes > options.budget_bytes: + eviction_output["error"] = "The tier is still over the budget, because the dumps of units not yet dropped are kept" + print(json.dumps(eviction_output)) + raise SystemExit(1) + print(json.dumps(eviction_output)) return tiers = list(ArchiveTier) if options.tier is None else [ArchiveTier(options.tier)] audited_objects = 0