From fe7394de14a28be3106e379e8a3fd9d0a0425843 Mon Sep 17 00:00:00 2001 From: PhiLily <252857470+PhiLily@users.noreply.github.com> Date: Mon, 5 Oct 2026 22:00:08 +0300 Subject: [PATCH] Preserve unreadable ticks and add targeted history removal Keep unreadable ticks and each schedule's newest readable tick during ordinary pruning. Add explicitly targeted unreadable-history removal, with dry-run listing and locked rechecks that preserve repaired rows. Stop dispatch workers and inspect a dry run before removal. History removal can repeat a tick or skip the next tick through re-anchoring. MySQL key filters use table collation; primary-key selection avoids equivalent-key matches and handles SQLite byte-valued keys. Delete tick rows without Django deletion signals on every pruning path. Preserve task signals and ordinary model deletion behavior. --- README.md | 2 +- docs/configuration.md | 20 +- docs/llms-full.txt | 33 +- docs/monitoring.md | 9 +- docs/recurring-tasks.md | 4 +- src/django_ox/_stored_read.py | 118 ++ src/django_ox/management/commands/ox_prune.py | 783 +++++++- tests/test_prune.py | 8 +- tests/test_prune_unreadable.py | 1642 +++++++++++++++++ tests/test_replica_reads.py | 61 +- tests/test_stored_read.py | 221 ++- 11 files changed, 2814 insertions(+), 87 deletions(-) create mode 100644 tests/test_prune_unreadable.py diff --git a/README.md b/README.md index 0d8e26d..a21f407 100644 --- a/README.md +++ b/README.md @@ -199,7 +199,7 @@ python manage.py ox_prune --older-than 7d | `--include-failed` | off | Also delete FAILED and LOST rows. By default they are kept: they hold the per-attempt tracebacks and can be retried. | | `--batch-size` | `1000` | Rows per DELETE statement, so pruning a large table never takes a long lock or builds a giant IN clause. | | `--dry-run` | off | Report how many rows would be deleted without deleting any. | -| `--format` | `text` | `json` prints one object on stdout instead of the two report lines: `queue`, `cutoff`, `statuses`, `task_rows`, `tick_rows` and `dry_run`. `queue` is `null` when no `--queue` is given, `cutoff` is ISO 8601, and `statuses` is a list. On a database error during deletion, the object is printed too, with counts of rows already deleted in committed batches, before the same non-zero exit. | +| `--format` | `text` | `json` prints one object on stdout; warnings stay on stderr. Ordinary keys: `queue`, `cutoff`, `statuses`, `task_rows`, `tick_rows`, `dry_run`, `unreadable_tick_rows`, `anchor_tick_rows`, `unreadable_ticks`. `queue` is `null` without `--queue`; `cutoff` is ISO 8601; `statuses` is a list. `unreadable_ticks` lists at most 100 `{pk, schedule_name, reason}` entries; counts cover all encountered unreadable rows, not a census. Purge keys: `purge_unreadable_ticks`, `schedule_keys`, `tick_pks`, `tick_rows`, `unreadable_ticks`, `kept_ticks`, `missing_tick_pks`, `dry_run`. Database errors during deletion still print committed-batch results before non-zero exit. | | `--database` | the alias `OxTask` writes to | Database alias to prune. The rows it reads and the rows it deletes are on that one alias. | Only SUCCESSFUL and DISCARDED rows (and, with `--include-failed`, FAILED and diff --git a/docs/configuration.md b/docs/configuration.md index f34eba2..71817c9 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -420,19 +420,17 @@ python manage.py ox_prune --older-than 7d | `--include-failed` | off | Also delete FAILED and LOST rows. By default they are kept, because they hold the per-attempt tracebacks and can be retried. | | `--batch-size` | `1000` | Rows per DELETE statement, so pruning a large table never takes a long lock or builds a giant IN clause. Must be at least 1. | | `--dry-run` | off | Report how many rows would be deleted without deleting any. | -| `--format` | `text` | `json` prints one object on stdout instead of the two report lines: `queue`, `cutoff`, `statuses`, `task_rows`, `tick_rows` and `dry_run`. `queue` is `null` when no `--queue` is given, `cutoff` is ISO 8601, and `statuses` is a list. On a database error during deletion, the object is printed too, with counts of rows already deleted in committed batches, before the same non-zero exit. | +| `--format` | `text` | `json` prints one object on stdout; warnings stay on stderr. Ordinary keys: `queue`, `cutoff`, `statuses`, `task_rows`, `tick_rows`, `dry_run`, `unreadable_tick_rows`, `anchor_tick_rows`, `unreadable_ticks`. `queue` is `null` without `--queue`; `cutoff` is ISO 8601; `statuses` is a list. `unreadable_ticks` lists at most 100 `{pk, schedule_name, reason}` entries; counts cover all encountered unreadable rows, not a census. Purge keys: `purge_unreadable_ticks`, `schedule_keys`, `tick_pks`, `tick_rows`, `unreadable_ticks`, `kept_ticks`, `missing_tick_pks`, `dry_run`. Database errors during deletion still print committed-batch results before non-zero exit. | | `--database` | the alias `OxTask` writes to | Database alias to prune. The rows it reads and the rows it deletes are on that one alias. | +| `--purge-unreadable-ticks` | off | Remove unreadable ticks targeted by `--schedule-key` or `--tick-pk`; a target is required. No schedule or task rows are removed. MySQL key matching follows the tick table's collation. Cannot combine with `--queue`, `--older-than` or `--include-failed`; `--batch-size`, `--database`, `--format` and `--dry-run` remain available. Stop affected dispatchers first. Preview with `--dry-run`, which lists rows without deleting them. | +| `--schedule-key KEY` | none | With `--purge-unreadable-ticks`, select unreadable ticks by settings schedule name or `db:` for a stored schedule. Repeatable. MySQL uses the tick table's collation, possibly matching case, accent or trailing-space variants. Inspect `--dry-run`, or target individual rows with `--tick-pk`. PostgreSQL and SQLite match exactly. | +| `--tick-pk PK` | none | With `--purge-unreadable-ticks`, target one tick by primary key. Repeatable. Use this when the schedule key cannot be read, including SQLite bytes. Readable rows are kept and reported. Missing or out-of-range keys report `No tick row N.` without preventing removal of other targets. | -Only SUCCESSFUL and DISCARDED rows (and, with `--include-failed`, FAILED and -LOST rows) whose `finished_at` is past the cutoff are deleted. READY, WAITING -and RUNNING rows are never touched, whatever their age. Rows from the -recurring-schedule tick -log are pruned with the same cutoff, always keeping each schedule's most -recent tick; that row anchors missed-tick recovery and deleting it would -make the schedule re-anchor. The latest tick row of a schedule that has -been removed from settings is kept by the same rule; such rows are -harmless and can be deleted by hand if unwanted. See -[Recurring tasks](recurring-tasks.md#missed-ticks). +Only SUCCESSFUL and DISCARDED task rows past the `finished_at` cutoff are deleted, plus FAILED and LOST rows with `--include-failed`. READY, WAITING and RUNNING rows are never touched. Ordinary tick pruning uses the same cutoff but preserves every unreadable tick and each schedule's newest readable tick, whatever its age. A tick that becomes newest during pruning is also preserved. + +This rule also preserves history for schedules removed from settings. Such rows are harmless; removing their history by hand is possible but can affect recovery. Preserved unreadable ticks produce warnings without failing ordinary pruning or `--dry-run`; database failures still exit non-zero. Warnings identify the first 100 encountered rows by primary key, key and reason. Counts describe encountered rows, not a complete census. See [Recurring tasks](recurring-tasks.md#missed-ticks). + +To remove unreadable ticks, stop the workers that dispatch the affected schedules. Preview with `ox_prune --purge-unreadable-ticks --schedule-key KEY --dry-run`, or use `--tick-pk PK` if the key cannot be read. Check the listed rows and the tasks those ticks recorded. Run the same command without `--dry-run`, then restart the workers. Removing history can repeat a tick or cause re-anchoring that skips the next tick. Tick deletion uses the command's own `DELETE` and sends no delete signals. `--queue` narrows the task rows, not the tick log. A run for one queue prunes every schedule's old ticks at its own cutoff, so when queues are diff --git a/docs/llms-full.txt b/docs/llms-full.txt index 792e2b0..8187f1d 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -1512,19 +1512,17 @@ python manage.py ox_prune --older-than 7d | `--include-failed` | off | Also delete FAILED and LOST rows. By default they are kept, because they hold the per-attempt tracebacks and can be retried. | | `--batch-size` | `1000` | Rows per DELETE statement, so pruning a large table never takes a long lock or builds a giant IN clause. Must be at least 1. | | `--dry-run` | off | Report how many rows would be deleted without deleting any. | -| `--format` | `text` | `json` prints one object on stdout instead of the two report lines: `queue`, `cutoff`, `statuses`, `task_rows`, `tick_rows` and `dry_run`. `queue` is `null` when no `--queue` is given, `cutoff` is ISO 8601, and `statuses` is a list. On a database error during deletion, the object is printed too, with counts of rows already deleted in committed batches, before the same non-zero exit. | +| `--format` | `text` | `json` prints one object on stdout; warnings stay on stderr. Ordinary keys: `queue`, `cutoff`, `statuses`, `task_rows`, `tick_rows`, `dry_run`, `unreadable_tick_rows`, `anchor_tick_rows`, `unreadable_ticks`. `queue` is `null` without `--queue`; `cutoff` is ISO 8601; `statuses` is a list. `unreadable_ticks` lists at most 100 `{pk, schedule_name, reason}` entries; counts cover all encountered unreadable rows, not a census. Purge keys: `purge_unreadable_ticks`, `schedule_keys`, `tick_pks`, `tick_rows`, `unreadable_ticks`, `kept_ticks`, `missing_tick_pks`, `dry_run`. Database errors during deletion still print committed-batch results before non-zero exit. | | `--database` | the alias `OxTask` writes to | Database alias to prune. The rows it reads and the rows it deletes are on that one alias. | +| `--purge-unreadable-ticks` | off | Remove unreadable ticks targeted by `--schedule-key` or `--tick-pk`; a target is required. No schedule or task rows are removed. MySQL key matching follows the tick table's collation. Cannot combine with `--queue`, `--older-than` or `--include-failed`; `--batch-size`, `--database`, `--format` and `--dry-run` remain available. Stop affected dispatchers first. Preview with `--dry-run`, which lists rows without deleting them. | +| `--schedule-key KEY` | none | With `--purge-unreadable-ticks`, select unreadable ticks by settings schedule name or `db:` for a stored schedule. Repeatable. MySQL uses the tick table's collation, possibly matching case, accent or trailing-space variants. Inspect `--dry-run`, or target individual rows with `--tick-pk`. PostgreSQL and SQLite match exactly. | +| `--tick-pk PK` | none | With `--purge-unreadable-ticks`, target one tick by primary key. Repeatable. Use this when the schedule key cannot be read, including SQLite bytes. Readable rows are kept and reported. Missing or out-of-range keys report `No tick row N.` without preventing removal of other targets. | -Only SUCCESSFUL and DISCARDED rows (and, with `--include-failed`, FAILED and -LOST rows) whose `finished_at` is past the cutoff are deleted. READY, WAITING -and RUNNING rows are never touched, whatever their age. Rows from the -recurring-schedule tick -log are pruned with the same cutoff, always keeping each schedule's most -recent tick; that row anchors missed-tick recovery and deleting it would -make the schedule re-anchor. The latest tick row of a schedule that has -been removed from settings is kept by the same rule; such rows are -harmless and can be deleted by hand if unwanted. See -[Recurring tasks](recurring-tasks.md#missed-ticks). +Only SUCCESSFUL and DISCARDED task rows past the `finished_at` cutoff are deleted, plus FAILED and LOST rows with `--include-failed`. READY, WAITING and RUNNING rows are never touched. Ordinary tick pruning uses the same cutoff but preserves every unreadable tick and each schedule's newest readable tick, whatever its age. A tick that becomes newest during pruning is also preserved. + +This rule also preserves history for schedules removed from settings. Such rows are harmless; removing their history by hand is possible but can affect recovery. Preserved unreadable ticks produce warnings without failing ordinary pruning or `--dry-run`; database failures still exit non-zero. Warnings identify the first 100 encountered rows by primary key, key and reason. Counts describe encountered rows, not a complete census. See [Recurring tasks](recurring-tasks.md#missed-ticks). + +To remove unreadable ticks, stop the workers that dispatch the affected schedules. Preview with `ox_prune --purge-unreadable-ticks --schedule-key KEY --dry-run`, or use `--tick-pk PK` if the key cannot be read. Check the listed rows and the tasks those ticks recorded. Run the same command without `--dry-run`, then restart the workers. Removing history can repeat a tick or cause re-anchoring that skips the next tick. Tick deletion uses the command's own `DELETE` and sends no delete signals. `--queue` narrows the task rows, not the tick log. A run for one queue prunes every schedule's old ticks at its own cutoff, so when queues are @@ -2065,9 +2063,7 @@ written the first time a worker saw the schedule; it enqueued nothing and marks the tick before the first real fire. A schedule-scoped failure rolls back its tick row and task, so the tick log alone does not report failed dispatches. -The recovery baseline is each schedule's most recent tick row, which is why -`ox_prune` always keeps it. See -[Configuration](configuration.md#ox_prune). +The recovery baseline is each schedule's most recent tick row. Ordinary `ox_prune` preserves every unreadable tick and each schedule's newest readable tick. Removing history can repeat work or cause re-anchoring. See [Configuration](configuration.md#ox_prune). --- @@ -6256,8 +6252,9 @@ The message text is not part of the contract. The keys are. | `schedule_dispatch_callback_failed` | WARNING | A `transaction.on_commit` callback registered by a `task_enqueued` receiver raised after the dispatch transaction committed. The task is enqueued, the tick is recorded and the dispatch is counted; the failure is the callback's. Carries `task_id`. | | `schedule_dispatch_failed` | WARNING | A dispatch pass was abandoned. `category=database` covers `django.db.Error`, including `InterfaceError`; `unexpected_exception` covers other unhandled exceptions. Failed rollback, a changed session or an unusable connection also abandons the pass. An unusable connection is dropped; claiming still runs. Dispatch is retried one schedule interval later. Each category gets a first traceback, then summaries at most once per minute. Carries `worker_id`, `database`, `category`, `error`, `failures` and `suppressed`. Marker reads and boundary heals retain their own events. | | `schedule_dispatch_resumed` | INFO | A dispatch pass completed after one or more abandoned passes. Logged once for the completed run of failures, covering both categories. Carries `worker_id`, `database` and `failures`, the total abandoned passes in that run. | -| `schedule_tick_unreadable` | WARNING | A schedule was skipped this pass because a tick in its history holds a value that cannot be read. Schedule skipped; tick history was not treated as absent: no tick is recorded and no task is enqueued for it, and the other schedules in the pass are dispatched. `phase` is `latest` for the newest tick within the pass's bound, or `anchor` for the earliest tick, which decides a settings-declared schedule's first sighting. The schedule stays skipped until the tick is repaired or removed. Logged once, then at most one summary per 60 seconds per schedule per worker. Carries `schedule`, `schedule_key`, `schedule_pk`, `phase`, `reason`, `tick_pk`, `database`, `failures` and `suppressed`. | +| `schedule_tick_unreadable` | WARNING | A schedule was skipped this pass because a tick in its history holds a value that cannot be read. Schedule skipped; tick history was not treated as absent: no tick is recorded and no task is enqueued for it, and the other schedules in the pass are dispatched. `phase` is `latest` for the newest tick within the pass's bound, or `anchor` for the earliest tick, which decides a settings-declared schedule's first sighting. The schedule stays skipped until the tick is repaired or removed. Logged once, then at most one summary per 60 seconds per schedule per worker. Carries `schedule`, `schedule_key`, `schedule_pk`, `phase`, `reason`, `tick_pk`, `database`, `failures` and `suppressed`. `ox_prune` logs the same event, with `surface` set to `prune`, when it met a schedule tick row it cannot read and kept it: one record per row for the first 100 rows of a run, carrying `surface`, `database`, `schedule_key`, `tick_pk` and `reason`, then one record carrying `unlisted` for the rest. From `ox_prune` it is logged only where a log handler is configured; the same rows are listed on stderr either way. | | `schedule_tick_readable` | INFO | This worker read a schedule's tick history again after skipping it with `schedule_tick_unreadable`. Logged once per run of skipped passes. Carries `worker_id`, `schedule`, `schedule_key`, `schedule_pk`, `database` and `failures`, the number of skipped passes in that run. | +| `schedule_tick_history_removed` | WARNING | `ox_prune --purge-unreadable-ticks` removed an unreadable tick row. Logged per row as each batch commits, with `database`, `schedule_key`, `tick_pk` and `reason`. Logging requires a `django_ox` handler; stdout lists removed rows regardless. Removing history can repeat a tick or cause re-anchoring that skips the next tick. | | `schedule_source_unavailable` | WARNING | The stored schedules or their change marker could not be read. A change marker holding a value that cannot be read is reported once, then at most once a minute while it remains unreadable. During that time, the worker reads the schedules in full at each reconcile interval instead of relying only on the last set it read. A database error on the marker read is reported on every dispatch pass while it lasts, and the worker keeps running on the last set it read. A database error during a full read stops the pass. A stream of this event right after a deploy can mean the code is running ahead of migration `0007`. | | `schedule_lock_unavailable` | WARNING | The database gave up waiting for a lock another worker held: a stored schedule's row, or the tick row a settings-declared schedule's dispatch claims. MySQL's lock-wait timeout or a deadlock it resolved against this worker, SQLite's busy timeout, PostgreSQL's `lock_timeout` or a deadlock. The schedule is skipped this pass and its tick fires on a later one if still unclaimed. No traceback; a stream can indicate long dispatch transactions, often from a slow `task_enqueued` receiver. On MySQL with three or more dispatchers, a refused settings schedule can also produce 1213 deadlock warnings when the claimant rolls back and workers queued on its tick key deadlock; read these beside `schedule_dispatch_error` for the same schedule. | | `schedule_boundary_healed` | INFO | A stored schedule's timing had changed without its activation boundary moving, so the boundary was moved onto the current timing. Expected after a bulk update; repeated for one schedule is not. | @@ -6355,9 +6352,11 @@ A failed connect while recording a stuck attempt logs | `reason` | `schedule_row_skipped` | Why a full or locked read rejected the row. Stored-value reasons name the field, quote its bounded value and explain the problem, including a maximum `boundary_generation`. Unregistered task keys are identified against this process's registry. | | `fields` | `schedule_row_skipped` | Fields rejected by the guarded reader, in model order, as a list. Includes a maximum `boundary_generation`. Empty for build failures, including unregistered task keys, or when the read cannot identify a field. | | `reason` | `schedule_tick_unreadable`, `schedule_tick_history_removed` | Why the tick could not be read. Names the column and quotes the stored value, escaped and shortened. | -| `schedule_key` | `schedule_tick_unreadable`, `schedule_tick_readable` | The key the schedule's ticks are recorded under: a settings-declared schedule's name, or `db:` for a stored schedule. | +| `schedule_key` | `schedule_tick_unreadable`, `schedule_tick_readable`, `schedule_tick_history_removed` | Tick-history key: a settings schedule's name or `db:` for a stored schedule. Prune records escape control characters and bound it to 200 characters; an unreadable key is empty. | | `phase` | `schedule_tick_unreadable` | Worker: `latest` for the newest tick within the pass's bound; `anchor` for a settings schedule's earliest tick. Admin: always `latest`. Absent from prune records. | -| `tick_pk` | `schedule_tick_unreadable` | The primary key of the tick row that could not be read, or `None` when it could not be determined. | +| `tick_pk` | `schedule_tick_unreadable`, `schedule_tick_history_removed` | Tick row's primary key. For worker `phase=latest`, looked up only when logging; `None` if missing or lookup fails. The schedule remains skipped either way. Anchor, admin and prune reads supply the key directly. | +| `surface` | `schedule_tick_unreadable` | What met the unreadable tick row: `prune` for `ox_prune`. Absent on a worker's record. | +| `unlisted` | `schedule_tick_unreadable` | On the last record of an `ox_prune` run that kept more than 100 unreadable tick rows: how many of them were not named one by one. | | `category` | `schedule_dispatch_failed` | What abandoned the pass: `database` or `unexpected_exception`. | | `queues` | `worker_started` | The worker's queues. | | `concurrency` | `worker_started`, `connection_pool_too_small` | The worker's task-thread concurrency, set by `--concurrency`. | diff --git a/docs/monitoring.md b/docs/monitoring.md index 3c3218b..7a51d3d 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -700,8 +700,9 @@ The message text is not part of the contract. The keys are. | `schedule_dispatch_callback_failed` | WARNING | A `transaction.on_commit` callback registered by a `task_enqueued` receiver raised after the dispatch transaction committed. The task is enqueued, the tick is recorded and the dispatch is counted; the failure is the callback's. Carries `task_id`. | | `schedule_dispatch_failed` | WARNING | A dispatch pass was abandoned. `category=database` covers `django.db.Error`, including `InterfaceError`; `unexpected_exception` covers other unhandled exceptions. Failed rollback, a changed session or an unusable connection also abandons the pass. An unusable connection is dropped; claiming still runs. Dispatch is retried one schedule interval later. Each category gets a first traceback, then summaries at most once per minute. Carries `worker_id`, `database`, `category`, `error`, `failures` and `suppressed`. Marker reads and boundary heals retain their own events. | | `schedule_dispatch_resumed` | INFO | A dispatch pass completed after one or more abandoned passes. Logged once for the completed run of failures, covering both categories. Carries `worker_id`, `database` and `failures`, the total abandoned passes in that run. | -| `schedule_tick_unreadable` | WARNING | A schedule was skipped this pass because a tick in its history holds a value that cannot be read. Schedule skipped; tick history was not treated as absent: no tick is recorded and no task is enqueued for it, and the other schedules in the pass are dispatched. `phase` is `latest` for the newest tick within the pass's bound, or `anchor` for the earliest tick, which decides a settings-declared schedule's first sighting. The schedule stays skipped until the tick is repaired or removed. Logged once, then at most one summary per 60 seconds per schedule per worker. Carries `schedule`, `schedule_key`, `schedule_pk`, `phase`, `reason`, `tick_pk`, `database`, `failures` and `suppressed`. | +| `schedule_tick_unreadable` | WARNING | A schedule was skipped this pass because a tick in its history holds a value that cannot be read. Schedule skipped; tick history was not treated as absent: no tick is recorded and no task is enqueued for it, and the other schedules in the pass are dispatched. `phase` is `latest` for the newest tick within the pass's bound, or `anchor` for the earliest tick, which decides a settings-declared schedule's first sighting. The schedule stays skipped until the tick is repaired or removed. Logged once, then at most one summary per 60 seconds per schedule per worker. Carries `schedule`, `schedule_key`, `schedule_pk`, `phase`, `reason`, `tick_pk`, `database`, `failures` and `suppressed`. `ox_prune` logs the same event, with `surface` set to `prune`, when it met a schedule tick row it cannot read and kept it: one record per row for the first 100 rows of a run, carrying `surface`, `database`, `schedule_key`, `tick_pk` and `reason`, then one record carrying `unlisted` for the rest. From `ox_prune` it is logged only where a log handler is configured; the same rows are listed on stderr either way. | | `schedule_tick_readable` | INFO | This worker read a schedule's tick history again after skipping it with `schedule_tick_unreadable`. Logged once per run of skipped passes. Carries `worker_id`, `schedule`, `schedule_key`, `schedule_pk`, `database` and `failures`, the number of skipped passes in that run. | +| `schedule_tick_history_removed` | WARNING | `ox_prune --purge-unreadable-ticks` removed an unreadable tick row. Logged per row as each batch commits, with `database`, `schedule_key`, `tick_pk` and `reason`. Logging requires a `django_ox` handler; stdout lists removed rows regardless. Removing history can repeat a tick or cause re-anchoring that skips the next tick. | | `schedule_source_unavailable` | WARNING | The stored schedules or their change marker could not be read. A change marker holding a value that cannot be read is reported once, then at most once a minute while it remains unreadable. During that time, the worker reads the schedules in full at each reconcile interval instead of relying only on the last set it read. A database error on the marker read is reported on every dispatch pass while it lasts, and the worker keeps running on the last set it read. A database error during a full read stops the pass. A stream of this event right after a deploy can mean the code is running ahead of migration `0007`. | | `schedule_lock_unavailable` | WARNING | The database gave up waiting for a lock another worker held: a stored schedule's row, or the tick row a settings-declared schedule's dispatch claims. MySQL's lock-wait timeout or a deadlock it resolved against this worker, SQLite's busy timeout, PostgreSQL's `lock_timeout` or a deadlock. The schedule is skipped this pass and its tick fires on a later one if still unclaimed. No traceback; a stream can indicate long dispatch transactions, often from a slow `task_enqueued` receiver. On MySQL with three or more dispatchers, a refused settings schedule can also produce 1213 deadlock warnings when the claimant rolls back and workers queued on its tick key deadlock; read these beside `schedule_dispatch_error` for the same schedule. | | `schedule_boundary_healed` | INFO | A stored schedule's timing had changed without its activation boundary moving, so the boundary was moved onto the current timing. Expected after a bulk update; repeated for one schedule is not. | @@ -799,9 +800,11 @@ A failed connect while recording a stuck attempt logs | `reason` | `schedule_row_skipped` | Why a full or locked read rejected the row. Stored-value reasons name the field, quote its bounded value and explain the problem, including a maximum `boundary_generation`. Unregistered task keys are identified against this process's registry. | | `fields` | `schedule_row_skipped` | Fields rejected by the guarded reader, in model order, as a list. Includes a maximum `boundary_generation`. Empty for build failures, including unregistered task keys, or when the read cannot identify a field. | | `reason` | `schedule_tick_unreadable`, `schedule_tick_history_removed` | Why the tick could not be read. Names the column and quotes the stored value, escaped and shortened. | -| `schedule_key` | `schedule_tick_unreadable`, `schedule_tick_readable` | The key the schedule's ticks are recorded under: a settings-declared schedule's name, or `db:` for a stored schedule. | +| `schedule_key` | `schedule_tick_unreadable`, `schedule_tick_readable`, `schedule_tick_history_removed` | Tick-history key: a settings schedule's name or `db:` for a stored schedule. Prune records escape control characters and bound it to 200 characters; an unreadable key is empty. | | `phase` | `schedule_tick_unreadable` | Worker: `latest` for the newest tick within the pass's bound; `anchor` for a settings schedule's earliest tick. Admin: always `latest`. Absent from prune records. | -| `tick_pk` | `schedule_tick_unreadable` | The primary key of the tick row that could not be read, or `None` when it could not be determined. | +| `tick_pk` | `schedule_tick_unreadable`, `schedule_tick_history_removed` | Tick row's primary key. For worker `phase=latest`, looked up only when logging; `None` if missing or lookup fails. The schedule remains skipped either way. Anchor, admin and prune reads supply the key directly. | +| `surface` | `schedule_tick_unreadable` | What met the unreadable tick row: `prune` for `ox_prune`. Absent on a worker's record. | +| `unlisted` | `schedule_tick_unreadable` | On the last record of an `ox_prune` run that kept more than 100 unreadable tick rows: how many of them were not named one by one. | | `category` | `schedule_dispatch_failed` | What abandoned the pass: `database` or `unexpected_exception`. | | `queues` | `worker_started` | The worker's queues. | | `concurrency` | `worker_started`, `connection_pool_too_small` | The worker's task-thread concurrency, set by `--concurrency`. | diff --git a/docs/recurring-tasks.md b/docs/recurring-tasks.md index 44d2021..30f1b1a 100644 --- a/docs/recurring-tasks.md +++ b/docs/recurring-tasks.md @@ -318,6 +318,4 @@ written the first time a worker saw the schedule; it enqueued nothing and marks the tick before the first real fire. A schedule-scoped failure rolls back its tick row and task, so the tick log alone does not report failed dispatches. -The recovery baseline is each schedule's most recent tick row, which is why -`ox_prune` always keeps it. See -[Configuration](configuration.md#ox_prune). +The recovery baseline is each schedule's most recent tick row. Ordinary `ox_prune` preserves every unreadable tick and each schedule's newest readable tick. Removing history can repeat work or cause re-anchoring. See [Configuration](configuration.md#ox_prune). diff --git a/src/django_ox/_stored_read.py b/src/django_ox/_stored_read.py index f07995f..52f5e5a 100644 --- a/src/django_ox/_stored_read.py +++ b/src/django_ox/_stored_read.py @@ -987,6 +987,124 @@ def newest_tick_pk( return None +@dataclass(frozen=True) +class TickRow: + """ + One tick row as `read_ticks` reads it: its primary key, its schedule + key, and `at` when its tick reads or `reason`, safe to print, when it + does not. A row whose schedule key is what does not read has `reason` + say so and an empty `key`: text the driver cannot hand over, or bytes + where the text should be, which SQLite keeps and hands over as they + are. No schedule has either for a key. + """ + + pk: int + key: str + at: datetime | None = None + reason: str | None = None + cause: BaseException | None = field(default=None, compare=False, repr=False) + + @property + def unreadable(self) -> bool: + return self.reason is not None + + +def read_ticks( + query: QuerySet[OxScheduleTick], *, using: str, limit: int | None = None +) -> list[TickRow]: + """ + The tick rows `query` selects, in its order, each answered on its own. + + `query` is the caller's selection, filtered and ordered as it needs, and + `limit` caps it. Only each row's key, schedule key and tick are fetched, + the tick as text, so a row's other columns cannot fail the read. + + A selection that fails as a whole, on a value the driver cannot hand + over, is read again as primary keys, which always read, and then those + rows in halves until the one that fails is alone. Those reads are made + only after a failure known to be a value's, and only while the + connection is known to take another statement (`can_read_on`): an error + the database raised is raised as it is, with nothing read after it. + """ + selected = query.annotate(ox_stored_scheduled_for=as_text("scheduled_for")) + fetch = selected.values_list("pk", "schedule_name", "ox_stored_scheduled_for") + try: + fetched = list(fetch if limit is None else fetch[:limit]) + except Exception as exc: + if not can_read_on(exc, using=using): + raise + keys = query.values_list("pk", flat=True) + pks = list(keys if limit is None else keys[:limit]) + found = _tick_rows(pks, using) + return [found[pk] for pk in pks if pk in found] + return [_tick_row(pk, key, text, using) for pk, key, text in fetched] + + +def _key_not_text(pk: int, key: object) -> TickRow: + """The answer for a tick row whose schedule key is not text.""" + return TickRow( + pk, "", reason=f"schedule_name {_NOT_TEXT.format(value=_shown(key))}" + ) + + +def _tick_row(pk: int, key: object, text: object, using: str) -> TickRow: + """One fetched tick row's answer.""" + if not isinstance(key, str): + return _key_not_text(pk, key) + try: + at = decode_datetime(text, using) + except UnreadableValue as exc: + return TickRow(pk, key, reason=f"scheduled_for {exc}", cause=exc) + return TickRow(pk, key, at=at) + + +def _tick_rows(pks: list[int], using: str) -> dict[int, TickRow]: + """These tick rows by key, or in halves when they do not read as a whole.""" + if not pks: + return {} + try: + fetched = list( + OxScheduleTick.objects.using(using) + .filter(pk__in=pks) + .annotate(ox_stored_scheduled_for=as_text("scheduled_for")) + .values_list("pk", "schedule_name", "ox_stored_scheduled_for") + ) + except Exception as exc: + if not can_read_on(exc, using=using): + raise + if len(pks) == 1: + return {pks[0]: _tick_alone(pks[0], using, exc)} + middle = len(pks) // 2 + return {**_tick_rows(pks[:middle], using), **_tick_rows(pks[middle:], using)} + return {pk: _tick_row(pk, key, text, using) for pk, key, text in fetched} + + +def _tick_alone(pk: int, using: str, cause: BaseException) -> TickRow: + """ + A tick row that failed to read on its own: its schedule key if that + reads, and the tick as what could not be handed over. + """ + try: + key = ( + OxScheduleTick.objects.using(using) + .filter(pk=pk) + .values_list("schedule_name", flat=True) + .first() + ) + except Exception as exc: + if not can_read_on(exc, using=using): + raise + return TickRow( + pk, + "", + reason=f"schedule_name {_NOT_FETCHED.format(error=_error_text(exc))}", + cause=exc, + ) + if key is not None and not isinstance(key, str): + return _key_not_text(pk, key) + return TickRow(pk, key or "", reason=_fetch_failed(cause), cause=cause) + + def earliest_tick( key: str, *, using: str, exclude: Collection[int] = (), lock: bool = False ) -> TickRead: diff --git a/src/django_ox/management/commands/ox_prune.py b/src/django_ox/management/commands/ox_prune.py index 7803d1f..29bbe8d 100644 --- a/src/django_ox/management/commands/ox_prune.py +++ b/src/django_ox/management/commands/ox_prune.py @@ -1,29 +1,252 @@ import json -from collections.abc import Sequence +import logging +from collections.abc import Iterator, Sequence +from dataclasses import dataclass, field, replace from datetime import UTC, datetime, timedelta -from typing import Any +from typing import Any, cast from django.core.management.base import CommandError, CommandParser from django.db import DatabaseError, connections, transaction -from django.db.models import F, QuerySet +from django.db.models import Exists, F, OuterRef, QuerySet from django.utils import timezone -from django_ox import _contention +from django_ox import _contention, stored +from django_ox._stored_read import TickRow, can_read_on, read_ticks from django_ox.durations import parse_duration from django_ox.management._database import DatabaseCommand from django_ox.models import OxScheduleTick, OxTask +logger = logging.getLogger("django_ox") + # Reserve a full day so conversion to any connection timezone cannot # underflow datetime.min. Apply the same floor on every engine so the # answer does not depend on the engine or connection timezone. _CUTOFF_FLOOR = datetime.min + timedelta(days=1) +#: --older-than when it is not given. +_DEFAULT_OLDER_THAN = "7d" + +#: The most unreadable ticks a run names one by one, in its warning, its log +#: and its JSON. The counts cover every one. +_NAMED_LIMIT = 100 + +# What the command says about tick rows it cannot read, all of it in this +# block. `key` is a schedule key as repr() quotes it, `pk` a tick row's +# primary key, `reason` why the row cannot be read, already safe to print. + +#: The flag without anything to remove. +_PURGE_NEEDS_A_TARGET = ( + "--purge-unreadable-ticks needs at least one --schedule-key or --tick-pk. On " + "MySQL, --schedule-key uses the tick table's collation, which may match keys " + "differing in case, accents or trailing spaces. Inspect --dry-run before removing " + "rows, or use --tick-pk to target individual rows." +) +#: A target without the flag that acts on it. +_TARGET_NEEDS_PURGE = "Use --schedule-key and --tick-pk with --purge-unreadable-ticks." +#: The flag beside options that prune. `flags` lists them, comma-separated. +_PURGE_COMBINED = ( + "--purge-unreadable-ticks cannot be combined with {flags}. Run ordinary pruning " + "separately." +) +#: One unreadable tick row, in a list. +_TICK_LINE = " tick {pk}, schedule {key}: {reason}" +#: The end of a list cut at _NAMED_LIMIT. +_MORE_LINE = " and {count} more." +#: Ordinary pruning met tick rows it cannot read and kept them. Then one +#: _TICK_LINE each, _MORE_LINE if the list was cut, and _UNREADABLE_REMEDY. +_UNREADABLE_KEPT = ( + "Warning: this run encountered {count} schedule tick row(s) it cannot read. All " + "were kept. Unreadable tick history can block automatic dispatch." +) +_UNREADABLE_REMEDY = ( + "To remove unreadable ticks, stop the workers that dispatch these schedules. Run " + "ox_prune --purge-unreadable-ticks with --schedule-key KEY or --tick-pk PK and " + "--dry-run. Use --tick-pk PK if the schedule key cannot be read. Check the listed " + "rows and the tasks those ticks recorded. Run the same command without --dry-run, " + "then start the workers again." +) +#: The stdout line beside the deletion counts. +_UNREADABLE_COUNT = ( + "{verb} {count} unreadable schedule tick row(s) encountered in this run." +) +#: Old ticks kept because each is the newest one its schedule has that reads. +_ANCHORS_COUNT = ( + "{verb} {count} schedule tick row(s) scheduled before {cutoff}, each the " + "newest readable tick of its schedule." +) +#: What --purge-unreadable-ticks removed, or would. Then one _TICK_LINE each. +_PURGE_COUNT = "{verb} {count} unreadable schedule tick row(s){colon}" +#: A tick named by --tick-pk whose scheduled time reads. +_NAMED_TICK_READS = ( + "Kept tick {pk}, schedule {key}: the row is readable, with scheduled_for {at}." +) +#: A tick that read when it came to be removed, repaired since it was found. +_TICK_REPAIRED = ( + "Kept tick {pk}, schedule {key}: the row is now readable, with scheduled_for {at}." +) +#: A tick named by --tick-pk that is not there. +_NO_SUCH_TICK = "No tick row {pk}." +#: Written whenever --purge-unreadable-ticks has something to remove. +_HISTORY_LOSS = ( + "Warning: removing dispatch history can make a schedule run a tick again, or " + "anchor again and skip its next tick. Stop the workers that dispatch these " + "schedules before removing history. Run with --dry-run first and check the listed " + "rows and the tasks those ticks recorded. Then run without --dry-run and start the " + "workers again." +) +#: The log line for each tick row ordinary pruning kept because it cannot be +#: read, and for those past _NAMED_LIMIT together. +_LOG_TICK_KEPT = "Tick %s, schedule key %s, cannot be read and was kept: %s" +_LOG_TICKS_KEPT = "%d more schedule tick rows cannot be read and were kept" +#: The log line for each tick row --purge-unreadable-ticks removed. +_LOG_TICK_REMOVED = "Removed tick %s, schedule key %s, because it could not be read: %s" + + +@dataclass +class _TickTally: + """What a pass over the tick log found and did, as far as it got.""" + + #: Tick rows deleted, or that a dry run would delete. + deleted: int = 0 + #: Tick rows past the cutoff kept because each was its schedule's newest + #: readable tick. + anchors: int = 0 + #: Tick rows kept because they cannot be read, by primary key. + unreadable: dict[int, TickRow] = field(default_factory=dict) + + def keep(self, row: TickRow) -> None: + """Count a tick row that cannot be read, once however often it is met.""" + # Without the exception that says why: it holds on to the whole + # batch the row was read in, for as long as the tally lives. + self.unreadable.setdefault(row.pk, replace(row, cause=None)) + + +def _named(row: TickRow) -> dict[str, Any]: + """An unreadable tick row as the JSON output names it.""" + return {"pk": row.pk, "schedule_name": row.key, "reason": row.reason} + + +def _chunks(values: Sequence[int], size: int) -> Iterator[list[int]]: + for start in range(0, len(values), size): + yield list(values[start : start + size]) + + +def _possible_keys(numbers: Sequence[int], alias: str) -> list[int]: + """ + Those of these numbers a tick row's primary key can be on this database. + + A number outside the column's range names no row. It is left out of the + read rather than sent: SQLite's driver refuses to bind an integer wider + than the column, and that refusal is not a row that does not read. + """ + low, high = connections[alias].ops.integer_field_range( + OxScheduleTick._meta.pk.get_internal_type() + ) + return [number for number in numbers if low <= number <= high] + + +def _delete_ticks(pks: list[int], alias: str) -> int: + """ + Delete tick rows by primary key alone, and say how many went. + + Django's delete() loads every row it deletes as soon as anything listens + for its delete signals, and a tick that cannot be read cannot be loaded. + So the rows go in one DELETE of this command's own, which reads nothing: + on the connection the command reads and writes through, the table and + its key quoted as that database quotes them, and the keys passed as + parameters, never more of them than the batch the caller is on. + + Every tick row this command removes goes this way, whether or not one + that cannot be read is among them, so no pre_delete or post_delete is + sent for a tick row it removes. Task rows are deleted by Django's + delete(), which sends theirs. + """ + if not pks: + return 0 + connection = connections[alias] + table = connection.ops.quote_name(OxScheduleTick._meta.db_table) + key = connection.ops.quote_name(cast(str, OxScheduleTick._meta.pk.column)) + placeholders = ", ".join(["%s"] * len(pks)) + with connection.cursor() as cursor: + cursor.execute( + f"DELETE FROM {table} WHERE {key} IN ({placeholders})", # noqa: S608 + pks, + ) + return int(cursor.rowcount) + + +def _older_than_their_anchor( + rows: list[TickRow], anchors: dict[str, int], alias: str +) -> set[int]: + """ + Which of these tick rows sort before the tick kept for their schedule, + that tick still being there. + + `anchors` is each schedule key's kept tick, by primary key. A row with + a tick of its schedule above it is not the schedule's newest, which is + all it takes for the row to go. The rows and their anchors are asked + for by primary key, in the database's own order, so no tick is read + and the cost is the batch's, however long a schedule's history is. + Rows of one schedule sort by their tick among themselves, whatever the + column's collation makes of two keys. + + A plain read, never a locking one: see `_newest_of_their_schedules`. + """ + wanted = {anchors[row.key] for row in rows if row.key in anchors} + if not wanted: + return set() + place = { + pk: index + for index, pk in enumerate( + OxScheduleTick.objects.using(alias) + .filter(pk__in=[*(row.pk for row in rows), *wanted]) + .order_by("schedule_name", "scheduled_for") + .values_list("pk", flat=True) + ) + } + older: set[int] = set() + for row in rows: + anchor = anchors.get(row.key) + if anchor is None: + continue + here, above = place.get(row.pk), place.get(anchor) + if here is not None and above is not None and here < above: + older.add(row.pk) + return older + + +def _newest_of_their_schedules(pks: list[int], alias: str) -> set[int]: + """ + Which of these tick rows is its schedule's newest tick right now. + + Asked of the database by key, in the database's own order, so no tick + is read: a row is its schedule's newest when no row of the same + schedule sorts after it. A plain read, never a locking one. A worker's + first-sighting read skips locked tick rows, and a schedule whose every + tick it skipped would look like one it had never seen. + + For the few rows `_older_than_their_anchor` cannot answer for. The + subquery runs once a row, and on MySQL each run walks the schedule's + history up to the row. + """ + if not pks: + return set() + ticks = OxScheduleTick.objects.using(alias) + newer = ticks.filter( + schedule_name=OuterRef("schedule_name"), + scheduled_for__gt=OuterRef("scheduled_for"), + ) + return set( + ticks.filter(pk__in=pks).filter(~Exists(newer)).values_list("pk", flat=True) + ) + class Command(DatabaseCommand): help = "Delete finished task rows older than a cutoff." _task_rows = 0 - _tick_rows = 0 + #: What the tick phase has found and done, set afresh by each run. + _ticks: _TickTally def add_arguments(self, parser: CommandParser) -> None: super().add_arguments(parser) @@ -34,11 +257,11 @@ def add_arguments(self, parser: CommandParser) -> None: ) parser.add_argument( "--older-than", - default="7d", + default=None, help=( "How long a task must have been finished before it is " "deleted. Forms: 7d, 24h, 90m, 45s, or a plain number of " - "seconds (default: %(default)s)." + f"seconds (default: {_DEFAULT_OLDER_THAN})." ), ) parser.add_argument( @@ -73,14 +296,76 @@ def add_arguments(self, parser: CommandParser) -> None: "(default: %(default)s)." ), ) + parser.add_argument( + "--purge-unreadable-ticks", + action="store_true", + help=( + "Remove unreadable tick rows targeted by --schedule-key or --tick-pk. " + "No schedule or task rows are removed. On MySQL, --schedule-key uses " + "the tick table's collation, which may match keys differing in case, " + "accents or trailing spaces. Stop the workers that dispatch these " + "schedules. Run with --dry-run first and check the listed rows and the " + "tasks those ticks recorded. Then run without --dry-run and start the " + "workers again. Removing history can repeat a tick or cause " + "re-anchoring that skips the next tick." + ), + ) + parser.add_argument( + "--schedule-key", + action="append", + default=None, + metavar="KEY", + help=( + "With --purge-unreadable-ticks, select unreadable ticks by schedule " + "key: a settings schedule's name, or db: for a stored schedule. " + "Repeatable. On MySQL, matching uses the tick table's collation, which " + "may include keys differing in case, accents or trailing spaces. " + "Inspect --dry-run, or use --tick-pk to target individual rows." + ), + ) + parser.add_argument( + "--tick-pk", + action="append", + type=int, + default=None, + metavar="PK", + help=( + "With --purge-unreadable-ticks, one tick row to remove by " + "primary key, if it cannot be read. Repeatable." + ), + ) def handle(self, *args: Any, **options: Any) -> None: if options["batch_size"] < 1: raise CommandError("--batch-size must be a positive integer.") + keys = list(dict.fromkeys(options["schedule_key"] or [])) + tick_pks = list(dict.fromkeys(options["tick_pk"] or [])) + if options["purge_unreadable_ticks"]: + # Asked for one thing by name, so nothing it was not asked for: + # a prune option beside it would otherwise do nothing, silently. + combined = [ + flag + for flag, given in ( + ("--queue", options["queue"] is not None), + ("--older-than", options["older_than"] is not None), + ("--include-failed", options["include_failed"]), + ) + if given + ] + if combined: + raise CommandError(_PURGE_COMBINED.format(flags=", ".join(combined))) + if not keys and not tick_pks: + raise CommandError(_PURGE_NEEDS_A_TARGET) + self._purge(self.database(options), keys, tick_pks, options) + return + if keys or tick_pks: + raise CommandError(_TARGET_NEEDS_PURGE) # Once, before the first statement. Every queryset below is built on # this alias, so the rows the command reads are the rows it deletes. alias = self.database(options) older_than = options["older_than"] + if older_than is None: + older_than = _DEFAULT_OLDER_THAN try: cutoff = timezone.now() - parse_duration(older_than) except OverflowError: @@ -116,29 +401,19 @@ def handle(self, *args: Any, **options: Any) -> None: prunable = prunable.filter(queue_name=queue) label = f"{label} (queue {queue})" - # Old schedule ticks are dispatch-log bookkeeping, but each - # schedule's latest tick is the anchor the dispatcher measures - # missed ticks against. Keep that row whatever its age: deleting it - # would make the schedule re-anchor and skip a pending tick. - ticks = OxScheduleTick.objects.using(alias) - anchors = [ - ticks.filter(schedule_name=name).latest("scheduled_for").pk - for name in ticks.values_list("schedule_name", flat=True).distinct() - ] - prunable_ticks = ticks.filter(scheduled_for__lt=cutoff).exclude(pk__in=anchors) - as_json = options["format"] == "json" + batch_size = options["batch_size"] + self._ticks = _TickTally() if options["dry_run"]: task_count = prunable.count() - tick_count = prunable_ticks.count() + self._prune_ticks(alias, cutoff, batch_size, write=False) if as_json: self._write_json( queue=queue, cutoff=cutoff, statuses=statuses, task_rows=task_count, - tick_rows=tick_count, dry_run=True, ) else: @@ -147,17 +422,16 @@ def handle(self, *args: Any, **options: Any) -> None: f"finished before {cutoff.isoformat()}." ) self.stdout.write( - f"Would delete {tick_count} schedule tick row(s) " + f"Would delete {self._ticks.deleted} schedule tick row(s) " f"scheduled before {cutoff.isoformat()}." ) + self._write_kept("Would keep", cutoff) + self._warn_unreadable(alias) return self._task_rows = 0 - self._tick_rows = 0 try: - deleted = self._delete_tasks_in_batches( - prunable, options["batch_size"], label, alias - ) + deleted = self._delete_tasks_in_batches(prunable, batch_size, label, alias) # Written before the ticks are touched, the way it was before # --format text gained a second deletion behind the same try. # A tick failure must not cost the operator the task count. @@ -166,11 +440,9 @@ def handle(self, *args: Any, **options: Any) -> None: f"Deleted {deleted} {label} task row(s) " f"finished before {cutoff.isoformat()}." ) - deleted_ticks = self._delete_in_batches( - prunable_ticks, options["batch_size"] - ) + self._prune_ticks(alias, cutoff, batch_size, write=True) except (CommandError, DatabaseError): - # DatabaseError too: _delete_in_batches has no retry wrapper, and + # DatabaseError too: the tick phase has no retry wrapper, and # is_contention() only knows PostgreSQL and MySQL codes, so a # lock on SQLite or a failure in the tick phase arrives raw. # Either way the committed rows are gone and their count is the @@ -181,7 +453,6 @@ def handle(self, *args: Any, **options: Any) -> None: cutoff=cutoff, statuses=statuses, task_rows=self._task_rows, - tick_rows=self._tick_rows, dry_run=False, ) raise @@ -192,15 +463,15 @@ def handle(self, *args: Any, **options: Any) -> None: cutoff=cutoff, statuses=statuses, task_rows=deleted, - tick_rows=deleted_ticks, dry_run=False, ) - return - - self.stdout.write( - f"Deleted {deleted_ticks} schedule tick row(s) " - f"scheduled before {cutoff.isoformat()}." - ) + else: + self.stdout.write( + f"Deleted {self._ticks.deleted} schedule tick row(s) " + f"scheduled before {cutoff.isoformat()}." + ) + self._write_kept("Kept", cutoff) + self._warn_unreadable(alias) def _write_json( self, @@ -209,9 +480,9 @@ def _write_json( cutoff: datetime, statuses: Sequence[str], task_rows: int, - tick_rows: int, dry_run: bool, ) -> None: + ticks = self._ticks self.stdout.write( json.dumps( { @@ -219,12 +490,434 @@ def _write_json( "cutoff": cutoff.isoformat(), "statuses": list(statuses), "task_rows": task_rows, - "tick_rows": tick_rows, + "tick_rows": ticks.deleted, + "unreadable_tick_rows": len(ticks.unreadable), + "anchor_tick_rows": ticks.anchors, + "unreadable_ticks": [ + _named(row) + for row in list(ticks.unreadable.values())[:_NAMED_LIMIT] + ], "dry_run": dry_run, } ) ) + def _write_kept(self, verb: str, cutoff: datetime) -> None: + """The report's lines for tick rows kept that the cutoff would not keep.""" + ticks = self._ticks + if ticks.unreadable: + self.stdout.write( + _UNREADABLE_COUNT.format(verb=verb, count=len(ticks.unreadable)) + ) + if ticks.anchors: + self.stdout.write( + _ANCHORS_COUNT.format( + verb=verb, count=ticks.anchors, cutoff=cutoff.isoformat() + ) + ) + + def _warn_unreadable(self, alias: str) -> None: + """ + Say on stderr, and in the log, which tick rows were kept because they + cannot be read. On stderr whatever the format, so a cron line sees + it and JSON on stdout stays one object. + """ + rows = list(self._ticks.unreadable.values()) + if not rows: + return + listed = rows[:_NAMED_LIMIT] + lines = [_UNREADABLE_KEPT.format(count=len(rows))] + lines += [ + _TICK_LINE.format(pk=row.pk, key=repr(row.key), reason=row.reason) + for row in listed + ] + if len(rows) > len(listed): + lines.append(_MORE_LINE.format(count=len(rows) - len(listed))) + lines.append(_UNREADABLE_REMEDY) + self.stderr.write("\n".join(lines), style_func=self.style.WARNING) + # Only to handlers someone configured. Without one, Python's last + # resort would print each record on stderr again, below the warning + # that already says it. + if not logger.hasHandlers(): + return + for row in listed: + logger.warning( + _LOG_TICK_KEPT, + row.pk, + repr(row.key), + row.reason, + extra={ + "event": "schedule_tick_unreadable", + "surface": "prune", + "database": alias, + # Escaped as the message beside it is: the key is + # whatever the tick row holds. + "schedule_key": stored._printable(row.key), + "tick_pk": row.pk, + "reason": row.reason, + }, + ) + if len(rows) > len(listed): + logger.warning( + _LOG_TICKS_KEPT, + len(rows) - len(listed), + extra={ + "event": "schedule_tick_unreadable", + "surface": "prune", + "database": alias, + "unlisted": len(rows) - len(listed), + }, + ) + + # -- the tick log ------------------------------------------------------ + + def _prune_ticks( + self, alias: str, cutoff: datetime, batch_size: int, *, write: bool + ) -> None: + """ + Delete tick rows scheduled before `cutoff`, keeping two kinds. + + Each schedule's newest tick that reads is the baseline the dispatcher + measures missed ticks against, and deleting it would make the + schedule anchor again and skip a pending tick, so it stays whatever + its age. A tick that cannot be read stays too, and is reported: what + it meant is unknown, and removing dispatch history on a guess can + run a tick twice or anchor over one. --purge-unreadable-ticks removes + those by name. + + No tick is loaded through Django's converters, which raise on a + value they cannot read and take the whole result with them. Ticks + are fetched as text and decoded one by one. The rows a cutoff + selects are visited a batch at a time in key order, since those kept + stay selected, and a batch's rows are checked once more, just before + they go, for any that has become its schedule's newest tick: because + the tick kept for its schedule has gone since, or because the row + itself was dispatched since. + + A row can hold a key the first read of the keys did not return. On + MySQL the column's collation takes two spellings of a key for one + key, by case, by an accent or by a trailing space, and returns one + of them for both; elsewhere it is a schedule first dispatched since + that read. The tick kept for such a key is looked up when its first + row is met, by the comparison the database itself makes, so the + rows of every spelling are checked a batch at a time like the rest. + + With `write` false, the same rows are chosen and counted, and + nothing is deleted. + """ + ticks = OxScheduleTick.objects.using(alias) + anchors: dict[str, int] = {} + for key in self._schedule_keys(alias, batch_size): + newest = self._newest_readable(key, alias, batch_size) + if newest is not None: + anchors[key] = newest + protected = set(anchors.values()) + # Keys met in a batch that turned out to have no tick that reads. + unanchored: set[str] = set() + tally = self._ticks + selected = ticks.filter(scheduled_for__lt=cutoff) + after = None + while True: + page = selected if after is None else selected.filter(pk__gt=after) + rows = read_ticks(page.order_by("pk"), using=alias, limit=batch_size) + if not rows: + break + after = rows[-1].pk + old: list[TickRow] = [] + for row in rows: + if row.unreadable: + tally.keep(row) + continue + if row.key not in anchors and row.key not in unanchored: + newest = self._newest_readable(row.key, alias, batch_size) + if newest is None: + unanchored.add(row.key) + else: + anchors[row.key] = newest + protected.add(newest) + if row.pk in protected: + tally.anchors += 1 + else: + old.append(row) + # Since the newest ticks were read, a schedule's newest may have + # gone, and an older one in this batch become its newest. Or a + # tick dispatched since is in this batch, above the one kept. + older = _older_than_their_anchor(old, anchors, alias) + newest_now = _newest_of_their_schedules( + [row.pk for row in old if row.pk not in older], alias + ) + tally.anchors += len(newest_now) + doomed = [row.pk for row in old if row.pk not in newest_now] + if doomed: + # Counted once deleted, so a failure leaves the count of what + # is gone. + tally.deleted += _delete_ticks(doomed, alias) if write else len(doomed) + if len(rows) < batch_size: + break + + def _schedule_keys(self, alias: str, batch_size: int) -> list[str]: + """ + Every schedule key the tick log holds. + + One statement, unless a key is text the driver cannot hand over, + which SQLite keeps. Then every tick row is read, a batch at a time + and each answered on its own. The keys that read are the answer, + and a row whose key does not read is kept and reported like a row + whose tick does not. + + SQLite also keeps bytes where a key should be, and hands them over + as they are, with no error. They are no schedule's key and are left + out; a row that holds them is kept and reported when the cutoff + selects it. + + The rows are read again that way only after that failure, which + is the driver's and leaves a caller's transaction as it was. A read + the database itself refuses is raised as it is, inside a caller's + transaction and with none open (`_stored_read.can_read_on`): its + error names no tick, and no savepoint is taken around a read of + the tick log to come back to. + """ + ticks = OxScheduleTick.objects.using(alias) + try: + return [ + key + for key in ticks.values_list("schedule_name", flat=True).distinct() + if isinstance(key, str) + ] + except Exception as exc: + if not can_read_on(exc, using=alias): + raise + keys: dict[str, None] = {} + after = None + while True: + page = ticks if after is None else ticks.filter(pk__gt=after) + rows = read_ticks(page.order_by("pk"), using=alias, limit=batch_size) + for row in rows: + if row.unreadable and not row.key: + self._ticks.keep(row) + else: + keys[row.key] = None + if len(rows) < batch_size: + return list(keys) + after = rows[-1].pk + + def _newest_readable(self, key: str, alias: str, batch_size: int) -> int | None: + """ + The primary key of `key`'s newest tick that reads, or None when none + does. The ticks above it, newer by the database's order and not + readable, are added to the tally as they are passed. + + Newest first, one row and then a batch at a time, so a schedule whose + newest tick reads costs one row. + """ + newest_first = ( + OxScheduleTick.objects.using(alias) + .filter(schedule_name=key) + .order_by("-scheduled_for") + ) + offset, size = 0, 1 + while True: + rows = read_ticks(newest_first[offset:], using=alias, limit=size) + for row in rows: + if not row.unreadable: + return row.pk + self._ticks.keep(row) + if len(rows) < size: + return None + offset += len(rows) + size = batch_size + + # -- --purge-unreadable-ticks -------------------------------------------- + + def _purge( + self, alias: str, keys: list[str], tick_pks: list[int], options: dict[str, Any] + ) -> None: + """ + Remove the named schedules' unreadable tick rows and the named + unreadable tick rows, and nothing else. + + What is removed is found first, and a dry run stops there. Each + batch is then locked, read again, and only the rows that still do + not read are removed: one repaired since it was found stays. Rows go + by primary key, in a transaction a batch. + """ + batch_size = options["batch_size"] + as_json = options["format"] == "json" + dry_run = options["dry_run"] + ticks = OxScheduleTick.objects.using(alias) + found: dict[int, TickRow] = {} + for key in keys: + after = None + while True: + page = ticks.filter(schedule_name=key) + if after is not None: + page = page.filter(pk__gt=after) + rows = read_ticks(page.order_by("pk"), using=alias, limit=batch_size) + for row in rows: + if row.unreadable: + found.setdefault(row.pk, row) + if len(rows) < batch_size: + break + after = rows[-1].pk + reads: list[TickRow] = [] + missing: list[int] = [] + for chunk in _chunks(tick_pks, batch_size): + named = { + row.pk: row + for row in read_ticks( + ticks.filter(pk__in=_possible_keys(chunk, alias)).order_by("pk"), + using=alias, + ) + } + for pk in chunk: + tick = named.get(pk) + if tick is None: + missing.append(pk) + elif tick.unreadable: + found.setdefault(pk, tick) + else: + reads.append(tick) + + removed: list[TickRow] = [] + repaired: list[TickRow] = [] + if dry_run: + removed = list(found.values()) + else: + try: + for chunk in _chunks(sorted(found), batch_size): + gone, kept = self._remove_unreadable(chunk, alias) + # Logged as each batch commits, so a failure later on + # leaves a record of every row already gone. + self._log_removed(gone, alias) + removed += gone + repaired += kept + except DatabaseError: + if as_json: + self._write_purge_json( + keys=keys, + tick_pks=tick_pks, + removed=removed, + kept=reads + repaired, + missing=missing, + dry_run=dry_run, + ) + raise + + if as_json: + self._write_purge_json( + keys=keys, + tick_pks=tick_pks, + removed=removed, + kept=reads + repaired, + missing=missing, + dry_run=dry_run, + ) + else: + verb = "Would remove" if dry_run else "Removed" + colon = ":" if removed else "." + lines = [_PURGE_COUNT.format(verb=verb, count=len(removed), colon=colon)] + lines += [ + _TICK_LINE.format(pk=row.pk, key=repr(row.key), reason=row.reason) + for row in removed + ] + lines += [ + _NAMED_TICK_READS.format(pk=row.pk, key=repr(row.key), at=_iso(row.at)) + for row in reads + ] + lines += [ + _TICK_REPAIRED.format(pk=row.pk, key=repr(row.key), at=_iso(row.at)) + for row in repaired + ] + lines += [_NO_SUCH_TICK.format(pk=pk) for pk in missing] + self.stdout.write("\n".join(lines)) + if found: + self.stderr.write(_HISTORY_LOSS, style_func=self.style.WARNING) + + def _remove_unreadable( + self, pks: list[int], alias: str + ) -> tuple[list[TickRow], list[TickRow]]: + """ + Remove those of these tick rows that still cannot be read, in one + transaction. Returns the rows removed and the rows that read now. + + The read again is the one that takes the rows' locks, so a repair + either committed before it, and is what it reads, or waits for this + transaction. Being a locking read, it reads the rows as they are + now, not as a snapshot older than this transaction. SQLite has no + row locks, and a transaction that reads first starts as a reader, + so there a no-op UPDATE makes this one the writer first. + """ + with transaction.atomic(using=alias): + held = OxScheduleTick.objects.using(alias).filter(pk__in=pks) + if connections[alias].features.has_select_for_update: + held = held.select_for_update() + else: + held.update(schedule_name=F("schedule_name")) + again = read_ticks(held.order_by("pk"), using=alias) + gone = [row for row in again if row.unreadable] + if gone: + _delete_ticks([row.pk for row in gone], alias) + # Returned once the block has committed. + return gone, [row for row in again if not row.unreadable] + + def _log_removed(self, removed: list[TickRow], alias: str) -> None: + """ + One record a removed tick row, every one of them, since each is + dispatch history gone for good. To the handlers someone configured: + without one, Python's last resort would print on stderr again what + stdout already lists. + """ + if not logger.hasHandlers(): + return + for row in removed: + logger.warning( + _LOG_TICK_REMOVED, + row.pk, + repr(row.key), + row.reason, + extra={ + "event": "schedule_tick_history_removed", + "database": alias, + "schedule_key": stored._printable(row.key), + "tick_pk": row.pk, + "reason": row.reason, + }, + ) + + def _write_purge_json( + self, + *, + keys: list[str], + tick_pks: list[int], + removed: list[TickRow], + kept: list[TickRow], + missing: list[int], + dry_run: bool, + ) -> None: + self.stdout.write( + json.dumps( + { + "purge_unreadable_ticks": True, + "schedule_keys": keys, + "tick_pks": tick_pks, + "tick_rows": len(removed), + "unreadable_ticks": [_named(row) for row in removed], + "kept_ticks": [ + { + "pk": row.pk, + "schedule_name": row.key, + "scheduled_for": _iso(row.at), + } + for row in kept + ], + "missing_tick_pks": missing, + "dry_run": dry_run, + } + ) + ) + + # -- task rows ----------------------------------------------------------- + def _delete_tasks_in_batches( self, prunable: QuerySet[OxTask], batch_size: int, label: str, alias: str ) -> int: @@ -305,12 +998,6 @@ def _delete_batch( # once the block has committed. return prunable.filter(pk__in=batch).delete()[0] - def _delete_in_batches(self, prunable: QuerySet[Any], batch_size: int) -> int: - deleted = 0 - while True: - batch = list(prunable.values_list("pk", flat=True)[:batch_size]) - if not batch: - break - deleted += prunable.filter(pk__in=batch).delete()[0] - self._tick_rows = deleted - return deleted + +def _iso(at: datetime | None) -> str | None: + return None if at is None else at.isoformat() diff --git a/tests/test_prune.py b/tests/test_prune.py index 4eeebf4..46c7ca4 100644 --- a/tests/test_prune.py +++ b/tests/test_prune.py @@ -286,7 +286,7 @@ def test_deletes_in_batches(self, django_assert_num_queries): # row SELECT, tick SET NULL, and DELETE. Inside the test's own # transaction that one is a savepoint, which adds its SAVEPOINT and # RELEASE. Then the empty terminating SELECT, and the tick pass adds - # its schedule-name SELECT and empty pk SELECT. + # its schedule-name SELECT and the empty read of its first batch. with django_assert_num_queries(24): out = prune("--batch-size=2") @@ -1153,6 +1153,9 @@ def test_json_format_prunes_and_outputs_exact_schema(self): assert data["statuses"] == ["SUCCESSFUL", "DISCARDED"] assert data["task_rows"] == 1 assert data["tick_rows"] == 1 + assert data["unreadable_tick_rows"] == 0 + assert data["anchor_tick_rows"] == 0 + assert data["unreadable_ticks"] == [] assert data["dry_run"] is False assert "cutoff" in data assert datetime.fromisoformat(data["cutoff"]) @@ -1162,6 +1165,9 @@ def test_json_format_prunes_and_outputs_exact_schema(self): "statuses", "task_rows", "tick_rows", + "unreadable_tick_rows", + "anchor_tick_rows", + "unreadable_ticks", "dry_run", } assert not OxTask.objects.filter(pk=old_ok.pk).exists() diff --git a/tests/test_prune_unreadable.py b/tests/test_prune_unreadable.py new file mode 100644 index 0000000..c1b01ad --- /dev/null +++ b/tests/test_prune_unreadable.py @@ -0,0 +1,1642 @@ +""" +ox_prune beside schedule tick rows that cannot be read. + +A tick written around django-ox, by SQL, a fixture or a restore, can hold a +scheduled time its column accepts and Django's read cannot convert: +PostgreSQL keeps year 10000, 'infinity' and '-infinity', SQLite keeps any +text, MySQL keeps zero dates. Django's read of such a value raises in the +middle of a result, or on SQLite can read as nothing at all. SQLite also +keeps a schedule key that is not UTF-8, which its driver refuses to hand +over at all. + +Ordinary pruning keeps every such tick and says so on stderr, keeps each +schedule's newest tick that reads, and prunes the rest by the cutoff as it +always did. `--purge-unreadable-ticks` removes the unreadable ticks it is +pointed at and nothing else. + +Every unreadable value is written by SQL, in the form the engine keeps it. +A test for another engine's value skips and says why. +""" + +import json +import logging +import os +import signal +import subprocess +import sys +import threading +from datetime import datetime, timedelta +from io import StringIO + +import pytest +from django.core.management import ManagementUtility, call_command +from django.core.management.base import CommandError +from django.db import ( + DEFAULT_DB_ALIAS, + DatabaseError, + DataError, + connection, + connections, + transaction, +) +from django.db.models import QuerySet +from django.db.models.signals import post_delete, pre_delete +from django.test.utils import CaptureQueriesContext +from django.utils import timezone + +from django_ox.models import OxScheduleTick, OxTask + +from .conftest import wait_for +from .contention import failing, simulated +from .test_schedule_isolation_worker import GUARD, finish, project, start +from .test_stored_read import Refusing + +TICKS = "django_ox_oxscheduletick" + +#: (engine, SQL literal, id): a tick each engine keeps that Django's read +#: raises on or misreads, and that sorts after every real date. +NEWEST = [ + pytest.param("postgresql", "'10000-01-01 00:00:00+00'", id="postgresql-year-10000"), + pytest.param("postgresql", "'infinity'", id="postgresql-infinity"), + pytest.param("sqlite", "'banana'", id="sqlite-banana"), + pytest.param("sqlite", "'9999-02-30 00:00:00'", id="sqlite-does-not-exist"), + pytest.param("mysql", "'9999-00-00 00:00:00'", id="mysql-zero-month"), +] + +#: The same, sorting before every real date. +OLDEST = [ + pytest.param("postgresql", "'-infinity'", id="postgresql-minus-infinity"), + pytest.param("postgresql", "'0001-01-01 00:00:00+00 BC'", id="postgresql-bc"), + pytest.param("sqlite", "'0000-02-30 00:00:00'", id="sqlite-year-zero"), + pytest.param("mysql", "'0000-00-00 00:00:00'", id="mysql-zero-date"), +] + +#: The same, sorting between the real dates around year 2020. PostgreSQL +#: keeps no such value: what it cannot hand over is out of range at one end. +BETWEEN = [ + pytest.param("sqlite", "'2020-02-30 00:00:00'", id="sqlite-does-not-exist"), + pytest.param("mysql", "'2020-00-10 00:00:00'", id="mysql-zero-month"), +] + +#: Text SQLite keeps and its driver cannot decode as UTF-8, as a schedule key. +UNDECODABLE_KEY = "CAST(X'6B80FF' AS TEXT)" + +#: Bytes SQLite keeps where a schedule key should be, and hands over as bytes. +BYTES_KEY = "X'6E696768746C79'" + +#: One of each for the engine this run is on. +NEWEST_HERE = { + "postgresql": "'infinity'", + "sqlite": "'9999-02-30 00:00:00'", + "mysql": "'9999-00-00 00:00:00'", +} +OLDEST_HERE = { + "postgresql": "'-infinity'", + "sqlite": "'0000-02-30 00:00:00'", + "mysql": "'0000-00-00 00:00:00'", +} + + +def by_sql(statement, params=(), *, using=None): + """ + Run one statement the way only SQL can write what it writes. + + MySQL refuses a zero date in the strict mode the suite runs in, so its + session is let off for the statement and put back. + """ + target = connection if using is None else using + with target.cursor() as cursor: + if target.vendor != "mysql": + cursor.execute(statement, params) + return + cursor.execute("SELECT @@SESSION.sql_mode") + (mode,) = cursor.fetchone() + cursor.execute("SET SESSION sql_mode = ''") + try: + cursor.execute(statement, params) + finally: + cursor.execute("SET SESSION sql_mode = %s", [mode]) + + +def insert_tick(key, literal, *, created=None): + """A tick row whose scheduled_for is `literal`, written by SQL; its key.""" + if created is None: + created = {"postgresql": "now()", "mysql": "NOW(6)"}.get( + connection.vendor, "'2026-10-05 00:00:00'" + ) + by_sql( + f"INSERT INTO {TICKS} (schedule_name, scheduled_for, task_id, created_at) " # noqa: S608 + f"VALUES (%s, {literal}, NULL, {created})", + [key], + ) + return ( + OxScheduleTick.objects.filter(schedule_name=key) + .order_by("-pk") + .values_list("pk", flat=True) + .first() + ) + + +def insert_tick_keyed(key_literal, *, days_ago): + """A tick row whose schedule key is `key_literal`, written by SQL; its key.""" + before = remaining() + when = connection.ops.adapt_datetimefield_value( + timezone.now() - timedelta(days=days_ago) + ) + by_sql( + f"INSERT INTO {TICKS} (schedule_name, scheduled_for, task_id, created_at) " # noqa: S608 + f"VALUES ({key_literal}, %s, NULL, %s)", + [when, when], + ) + (pk,) = remaining() - before + return pk + + +def make_tick(name, *, days_ago): + when = timezone.now() - timedelta(days=days_ago) + return OxScheduleTick.objects.create( + schedule_name=name, scheduled_for=when, created_at=when + ).pk + + +def only_on(vendor, why="kept by that engine only"): + if connection.vendor != vendor: + pytest.skip(f"a {vendor} value: {why}; this run is on {connection.vendor}") + + +def prune(*args): + out, err = StringIO(), StringIO() + call_command("ox_prune", *args, stdout=out, stderr=err) + return out.getvalue(), err.getvalue() + + +def remaining(): + return set(OxScheduleTick.objects.values_list("pk", flat=True)) + + +def _sql(sql): + return sql.replace('"', "").replace("`", "").upper() + + +def in_another_connection(statement, params=()): + """Run and commit one statement on a connection of its own.""" + other = connections.create_connection(DEFAULT_DB_ALIAS) + try: + by_sql(statement, params, using=other) + finally: + other.close() + + +@pytest.mark.django_db +class TestOrdinaryPruningKeepsWhatItCannotRead: + @pytest.mark.parametrize(("vendor", "literal"), NEWEST) + def test_an_unreadable_newest_tick_and_the_newest_one_that_reads_are_kept( + self, vendor, literal + ): + only_on(vendor) + make_tick("k", days_ago=30) + make_tick("k", days_ago=20) + newest_readable = make_tick("k", days_ago=10) + bad = insert_tick("k", literal) + + out, err = prune() + + assert remaining() == {newest_readable, bad} + assert "Deleted 2 schedule tick row(s)" in out + assert "Kept 1 unreadable schedule tick row(s) encountered in this run." in out + assert "Kept 1 schedule tick row(s) scheduled before " in out + assert f" tick {bad}, schedule 'k': scheduled_for " in err + assert "--purge-unreadable-ticks" in err + + @pytest.mark.parametrize(("vendor", "literal"), OLDEST) + def test_an_unreadable_oldest_tick_is_kept(self, vendor, literal): + only_on(vendor) + bad = insert_tick("k", literal) + make_tick("k", days_ago=30) + newest = make_tick("k", days_ago=10) + + out, err = prune() + + assert remaining() == {bad, newest} + assert "Deleted 1 schedule tick row(s)" in out + assert f" tick {bad}, schedule 'k': scheduled_for " in err + + @pytest.mark.parametrize(("vendor", "literal"), BETWEEN) + def test_an_unreadable_tick_between_readable_ones_is_kept(self, vendor, literal): + only_on(vendor) + make_tick("k", days_ago=3000) + bad = insert_tick("k", literal) + make_tick("k", days_ago=30) + newest = make_tick("k", days_ago=10) + + out, err = prune() + + assert remaining() == {bad, newest} + assert "Deleted 2 schedule tick row(s)" in out + assert "Kept 1 unreadable schedule tick row(s) encountered in this run." in out + assert f" tick {bad}, schedule 'k': scheduled_for " in err + + @pytest.mark.parametrize(("vendor", "literal"), OLDEST + NEWEST) + def test_an_only_tick_that_cannot_be_read_is_kept(self, vendor, literal): + only_on(vendor) + bad = insert_tick("settings-every-minute", literal) + + out, err = prune() + + assert remaining() == {bad} + assert "Deleted 0 schedule tick row(s)" in out + assert "Kept 1 unreadable schedule tick row(s) encountered in this run." in out + assert f" tick {bad}, schedule 'settings-every-minute': " in err + + def test_a_key_no_schedule_has_is_pruned_by_the_same_rules(self): + make_tick("db:999999", days_ago=30) + newest_readable = make_tick("db:999999", days_ago=20) + bad = insert_tick("db:999999", NEWEST_HERE[connection.vendor]) + + out, err = prune() + + assert remaining() == {newest_readable, bad} + assert "Deleted 1 schedule tick row(s)" in out + assert f" tick {bad}, schedule 'db:999999': " in err + + def test_a_tick_whose_created_at_cannot_be_read_prunes_without_being_loaded(self): + # Django's delete() loads each row it deletes once anything listens + # for its signals. This tick's scheduled time reads and is old, so + # it goes; its creation time does not read, so loading it would + # fail. + literal = { + "postgresql": "'10000-01-01 00:00:00+00'", + "sqlite": "'9999-02-30 00:00:00'", + "mysql": "'0000-00-00 00:00:00'", + }[connection.vendor] + old = timezone.now() - timedelta(days=30) + by_sql( + f"INSERT INTO {TICKS} (schedule_name, scheduled_for, task_id, created_at) " # noqa: S608 + f"VALUES (%s, %s, NULL, {literal})", + ["k", connection.ops.adapt_datetimefield_value(old)], + ) + newest = make_tick("k", days_ago=1) + + def listener(sender, **kwargs): + pass + + pre_delete.connect(listener, weak=False, dispatch_uid="test_prune_listener") + try: + out, _ = prune() + finally: + pre_delete.disconnect(dispatch_uid="test_prune_listener") + + assert remaining() == {newest} + assert "Deleted 1 schedule tick row(s)" in out + + def test_dry_run_chooses_what_the_run_deletes_and_writes_nothing(self): + newest = NEWEST_HERE[connection.vendor] + oldest = OLDEST_HERE[connection.vendor] + insert_tick("a", newest) + for days in (30, 20, 10): + make_tick("a", days_ago=days) + insert_tick("b", oldest) + make_tick("b", days_ago=30) + make_tick("b", days_ago=1) + insert_tick("c", oldest) + before = remaining() + + dry = json.loads(prune("--dry-run", "--format", "json")[0]) + + assert remaining() == before + done = json.loads(prune("--format", "json")[0]) + for figure in ("tick_rows", "unreadable_tick_rows", "anchor_tick_rows"): + assert dry[figure] == done[figure], figure + + def named(data): + return sorted(data["unreadable_ticks"], key=lambda named: named["pk"]) + + assert named(dry) == named(done) + assert (dry["tick_rows"], dry["unreadable_tick_rows"]) == (3, 3) + assert dry["anchor_tick_rows"] == 1 + assert dry["dry_run"] is True + assert len(remaining()) == len(before) - 3 + + def test_dry_run_text_says_what_it_would_keep(self): + bad = insert_tick("k", OLDEST_HERE[connection.vendor]) + make_tick("k", days_ago=30) + + out, err = prune("--dry-run") + + assert "Would delete 0 schedule tick row(s)" in out + assert ( + "Would keep 1 unreadable schedule tick row(s) encountered in this run." + in out + ) + assert "Would keep 1 schedule tick row(s) scheduled before " in out + assert f" tick {bad}, schedule 'k': " in err + + def test_json_is_one_object_on_stdout_and_the_warning_goes_to_stderr(self): + make_tick("k", days_ago=30) + make_tick("k", days_ago=10) + bad = insert_tick("k", NEWEST_HERE[connection.vendor]) + + out, err = prune("--format", "json") + + data = json.loads(out) + assert out.strip() == json.dumps(data) + assert data["tick_rows"] == 1 + assert data["unreadable_tick_rows"] == 1 + assert data["anchor_tick_rows"] == 1 + ((named,),) = [data["unreadable_ticks"]] + assert named["pk"] == bad + assert named["schedule_name"] == "k" + assert named["reason"].startswith("scheduled_for ") + assert "Warning" not in out + assert err.startswith( + "Warning: this run encountered 1 schedule tick row(s) it cannot" + ) + assert f" tick {bad}, schedule 'k': " in err + + def test_from_the_command_line_it_completes_and_warns(self, capsys): + bad = insert_tick("k", OLDEST_HERE[connection.vendor]) + argv = ["manage.py", "ox_prune", "--format", "json", "--skip-checks"] + + # Returns, rather than ending in SystemExit as a failed command does. + ManagementUtility(argv).execute() + + captured = capsys.readouterr() + assert json.loads(captured.out)["unreadable_tick_rows"] == 1 + assert f" tick {bad}, schedule 'k': " in captured.err + assert "Traceback" not in captured.err + + def test_many_are_counted_and_the_first_hundred_named(self): + literal = NEWEST_HERE[connection.vendor] + bad = [insert_tick(f"k{i:03}", literal) for i in range(103)] + + out, err = prune("--format", "json") + + data = json.loads(out) + assert data["unreadable_tick_rows"] == 103 + assert len(data["unreadable_ticks"]) == 100 + assert {named["pk"] for named in data["unreadable_ticks"]} < set(bad) + assert err.count("\n tick ") == 100 + assert "\n and 3 more.\n" in err + assert remaining() == set(bad) + + def test_each_kept_tick_is_logged_by_event(self, caplog): + bad = insert_tick("k", OLDEST_HERE[connection.vendor]) + + with caplog.at_level(logging.WARNING, logger="django_ox"): + prune() + + (record,) = [ + r + for r in caplog.records + if getattr(r, "event", None) == "schedule_tick_unreadable" + ] + assert record.surface == "prune" + assert record.database == DEFAULT_DB_ALIAS + assert record.schedule_key == "k" + assert record.tick_pk == bad + assert record.reason.startswith("scheduled_for ") + + def test_a_key_holding_a_line_break_is_logged_escaped(self, caplog): + # A tick row's key is whatever SQL wrote there, and a record's + # `schedule_key` is printed by any format that names it. + key = "k\nCRITICAL forged\x1b[31m" + bad = insert_tick(key, OLDEST_HERE[connection.vendor]) + + with caplog.at_level(logging.WARNING, logger="django_ox"): + prune() + prune("--purge-unreadable-ticks", "--tick-pk", str(bad)) + + kept, removed = ( + next(r for r in caplog.records if getattr(r, "event", None) == event) + for event in ("schedule_tick_unreadable", "schedule_tick_history_removed") + ) + for record in (kept, removed): + assert record.tick_pk == bad + assert record.schedule_key == "k\\nCRITICAL forged\\x1b[31m" + assert record.schedule_key.isprintable() + assert record.getMessage().isprintable() + assert remaining() == set() + + def test_inside_a_callers_transaction_the_transaction_works_after(self): + make_tick("k", days_ago=30) + make_tick("k", days_ago=10) + insert_tick("k", NEWEST_HERE[connection.vendor]) + + with transaction.atomic(): + with CaptureQueriesContext(connection) as queries: + out, _ = prune() + assert OxTask.objects.count() == 0 + assert OxScheduleTick.objects.filter(schedule_name="k").count() == 2 + + assert "Deleted 1 schedule tick row(s)" in out + # No read of the tick log takes a savepoint in the caller's + # transaction, and with no task row to delete nothing else does. + assert not [ + q["sql"] for q in queries.captured_queries if "SAVEPOINT" in q["sql"] + ] + + @pytest.mark.parametrize( + "args", + [(), ("--dry-run",), ("--purge-unreadable-ticks", "--schedule-key", "k")], + ids=["prune", "dry-run", "purge"], + ) + def test_inside_a_callers_transaction_a_refused_read_is_raised_as_it_was( + self, args + ): + """ + A read of the tick log the database itself refuses ends a + PostgreSQL transaction for every statement after it, and the + command takes no savepoint to go back to. It ends with the + database's own error, nothing read after it and nothing deleted, + and the caller's transaction, rolled back, takes the caller's own + writes with it. On PostgreSQL and MySQL the server refuses the read + for real; SQLite has no statement it refuses that way, and the + error is raised in the read's place. + """ + make_tick("k", days_ago=30) + make_tick("k", days_ago=10) + insert_tick("k", NEWEST_HERE[connection.vendor]) + before = remaining() + refusing = Refusing(TICKS) + + with ( + pytest.raises(DataError) as caught, + transaction.atomic(), + connection.execute_wrapper(refusing), + ): + make_tick("the-callers-own", days_ago=1) + prune(*args) + + refusing.assert_raised_as_it_was(caught) + assert remaining() == before + + @pytest.mark.parametrize( + "literal", + [ + pytest.param("CAST(X'62616480FF' AS TEXT)", id="sorts-newest"), + pytest.param("CAST(X'30303030FF' AS TEXT)", id="sorts-oldest"), + ], + ) + @pytest.mark.parametrize("position", [0, 4, 8], ids=["first", "middle", "last"]) + def test_a_tick_the_driver_cannot_hand_over_is_kept(self, literal, position): + # Eight readable old ticks over four schedules, each schedule also + # with a young newest one, and the bad tick written at `position` + # among the old ones: one batch of nine, read whole or in halves. + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + old = [(f"k{i % 4}", 30 - i) for i in range(8)] + bad_key = "k1" + bad = None + for index, (key, days) in enumerate(old): + if index == position: + bad = insert_tick(bad_key, literal) + make_tick(key, days_ago=days) + if bad is None: + bad = insert_tick(bad_key, literal) + newest = {make_tick(f"k{i}", days_ago=1) for i in range(4)} + + out, err = prune("--batch-size", "9") + + assert remaining() == newest | {bad} + assert "Deleted 8 schedule tick row(s)" in out + assert f" tick {bad}, schedule '{bad_key}': scheduled_for could not" in err + assert "UTF-8" in err + + @pytest.mark.parametrize("args", [(), ("--dry-run",)], ids=["run", "dry-run"]) + def test_a_tick_whose_schedule_key_cannot_be_handed_over_is_kept(self, args): + # One old and one young, among two schedules' ordinary history. The + # read of every schedule's key is the first the tick log gets, and + # on these two rows it fails as a whole. + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + old = [make_tick(key, days_ago=30) for key in ("j", "k")] + newest = {make_tick(key, days_ago=1) for key in ("j", "k")} + bad = { + insert_tick_keyed(UNDECODABLE_KEY, days_ago=30), + insert_tick_keyed(UNDECODABLE_KEY, days_ago=1), + } + before = remaining() + + out, err = prune("--batch-size", "2", *args) + + verb, kept = ("Would delete", "Would keep") if args else ("Deleted", "Kept") + assert remaining() == (before if args else newest | bad) + assert remaining() >= newest | bad + assert f"{verb} {len(old)} schedule tick row(s)" in out + assert ( + f"{kept} 2 unreadable schedule tick row(s) encountered in this run." in out + ) + for pk in bad: + assert f" tick {pk}, schedule '': schedule_name could not be read" in err + + @pytest.mark.parametrize( + "args", + [(), ("--dry-run",), ("--format", "json")], + ids=["run", "dry-run", "json"], + ) + def test_a_tick_whose_schedule_key_is_not_text_is_kept(self, args): + # Two rows under bytes for a key, both old and both with a tick that + # reads, and above them one whose tick does not. The driver hands + # the bytes over without an error, so nothing fails on these rows. + only_on("sqlite", "SQLite keeps bytes where a schedule key should be") + old = make_tick("k", days_ago=30) + newest = make_tick("k", days_ago=1) + kept = { + insert_tick_keyed(BYTES_KEY, days_ago=30), + insert_tick_keyed(BYTES_KEY, days_ago=20), + } + by_sql( + f"INSERT INTO {TICKS} (schedule_name, scheduled_for, task_id, created_at) " # noqa: S608 + f"VALUES ({BYTES_KEY}, 'banana', NULL, '2026-10-05 00:00:00')" + ) + before = remaining() + + out, err = prune(*args) + + dry = "--dry-run" in args + assert remaining() == (before if dry else before - {old}) + assert newest in remaining() + for pk in kept: + assert ( + f" tick {pk}, schedule '': schedule_name holds b'nightly', " + "which is not text" + ) in err + if "json" in args: + data = json.loads(out) + assert out.strip() == json.dumps(data) + assert data["tick_rows"] == 1 + named = {row["pk"]: row for row in data["unreadable_ticks"]} + assert set(named) == kept + assert {row["schedule_name"] for row in named.values()} == {""} + else: + verb = "Would delete" if dry else "Deleted" + assert f"{verb} 1 schedule tick row(s)" in out + + def test_text_carrying_an_offset_with_use_tz_off_is_kept(self, settings): + only_on("sqlite", "SQLite keeps text with an offset whatever USE_TZ says") + if settings.USE_TZ: + pytest.skip("read with USE_TZ off; this run has it on") + make_tick("k", days_ago=30) + newest_readable = make_tick("k", days_ago=10) + bad = insert_tick("k", "'9999-01-01 00:00:00+00:00'") + + _, err = prune() + + assert remaining() == {newest_readable, bad} + assert f" tick {bad}, schedule 'k': scheduled_for " in err + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize( + "args", + [(), ("--dry-run",), ("--purge-unreadable-ticks", "--schedule-key", "k")], + ids=["prune", "dry-run", "purge"], +) +def test_with_no_transaction_open_a_refused_read_ends_the_command_as_it_was(args): + """ + Run as an operator runs it, with no transaction open, the command does + not read the tick log again more narrowly after a read the database + itself refused, though nothing would be in the way. That error says + nothing of one tick, so no tick is kept or named for it: the command + ends with the database's own error, nothing read after it and nothing + deleted. + """ + make_tick("k", days_ago=30) + make_tick("k", days_ago=10) + insert_tick("k", NEWEST_HERE[connection.vendor]) + before = remaining() + refusing = Refusing(TICKS) + + assert connection.get_autocommit() + assert not connection.in_atomic_block + with pytest.raises(DataError) as caught, connection.execute_wrapper(refusing): + prune(*args) + + refusing.assert_raised_as_it_was(caught) + assert remaining() == before + + +@pytest.mark.django_db(transaction=True) +class TestAnchorsUnderConcurrentChange: + def test_a_tick_that_became_its_schedules_newest_since_the_read_is_kept(self): + # Between the read that found each schedule's newest tick and the + # delete, that tick goes, deleted on another connection. The older + # one the batch was going to delete is now the schedule's newest, + # and deleting it as well would leave the schedule no history. + oldest = make_tick("k", days_ago=30) + middle = make_tick("k", days_ago=20) + newest = make_tick("k", days_ago=10) + state = {"selected": False, "done": False} + + def wrapper(execute, sql, params, many, context): + if state["selected"] and not state["done"]: + state["done"] = True + in_another_connection( + f"DELETE FROM {TICKS} WHERE id = %s", # noqa: S608 + [newest], + ) + result = execute(sql, params, many, context) + text = _sql(sql) + if ( + text.lstrip().startswith("SELECT") + and f"{TICKS.upper()}.SCHEDULED_FOR <" in text + ): + state["selected"] = True + return result + + with connection.execute_wrapper(wrapper): + out, _ = prune() + + assert state["done"], "the prune never read a batch" + assert remaining() == {middle} + assert oldest not in remaining() + assert "Deleted 1 schedule tick row(s)" in out + + def test_a_tick_dispatched_during_the_prune_is_kept(self): + # A worker dispatches while the prune is going through its batches, + # late, for an instant before the cutoff: the new tick is selected + # like any old one. It is its schedule's newest, the tick the next + # pass compares against, and deleting it would let that pass run + # the same instant again. + make_tick("k", days_ago=30) + make_tick("k", days_ago=20) + selection_newest = make_tick("k", days_ago=10) + late = timezone.now() - timedelta(hours=1) + state = {"selected": False, "done": False} + + def wrapper(execute, sql, params, many, context): + if state["selected"] and not state["done"]: + state["done"] = True + in_another_connection( + f"INSERT INTO {TICKS} " # noqa: S608 + "(schedule_name, scheduled_for, task_id, created_at) " + "VALUES ('k', %s, NULL, %s)", + [connection.ops.adapt_datetimefield_value(late)] * 2, + ) + result = execute(sql, params, many, context) + text = _sql(sql) + if ( + text.lstrip().startswith("SELECT") + and f"{TICKS.upper()}.SCHEDULED_FOR <" in text + ): + state["selected"] = True + return result + + with connection.execute_wrapper(wrapper): + out, _ = prune("--older-than", "0", "--batch-size", "1") + + assert state["done"], "the prune never read a batch" + dispatched = OxScheduleTick.objects.get(scheduled_for=late).pk + assert remaining() == {selection_newest, dispatched} + assert "Deleted 2 schedule tick row(s)" in out + + def test_a_prune_holds_no_lock_on_the_tick_it_keeps(self): + # A worker's first-sighting read is a locking read that skips locked + # tick rows. A prune whose rows are still locked, inside a caller's + # transaction, must leave that read the tick it keeps: had it locked + # that one too, the schedule would look like one never seen. + if not connection.features.has_select_for_update_skip_locked: + pytest.skip(f"{connection.vendor} has no SKIP LOCKED") + make_tick("k", days_ago=30) + make_tick("k", days_ago=20) + kept = make_tick("k", days_ago=10) + pruned = threading.Event() + release = threading.Event() + errors = [] + + def prune_and_hold(): + try: + with transaction.atomic(): + prune("--older-than", "0") + pruned.set() + release.wait(30) + except BaseException as exc: + errors.append(exc) + finally: + pruned.set() + connections.close_all() + + thread = threading.Thread(target=prune_and_hold) + thread.start() + try: + assert pruned.wait(30) + assert not errors, errors + with transaction.atomic(): + earliest = ( + OxScheduleTick.objects.filter(schedule_name="k") + .order_by("scheduled_for") + .select_for_update(skip_locked=True) + .values_list("pk", flat=True) + .first() + ) + finally: + release.set() + thread.join(60) + assert not errors, errors + assert earliest == kept + assert remaining() == {kept} + + +@pytest.mark.django_db(transaction=True) +def test_the_tick_pass_reads_and_deletes_a_batch_at_a_time(): + # Outside any transaction, so each read is its statement alone. The + # schedule-name read, one read a schedule for its newest tick, then for + # each batch its read, the check for any that became its schedule's + # newest, and the DELETE. The last batch is short, so no empty read + # follows it. + for key in ("a", "b", "c"): + make_tick(key, days_ago=30) + make_tick(key, days_ago=20) + make_tick(key, days_ago=1) + + with CaptureQueriesContext(connection) as queries: + out, _ = prune("--batch-size", "4") + + statements = [_sql(q["sql"]).lstrip() for q in queries.captured_queries] + ticks = [s for s in statements if TICKS.upper() in s] + selects = [s for s in ticks if s.startswith("SELECT")] + deletes = [s for s in ticks if s.startswith("DELETE")] + assert len(selects) == 1 + 3 + 2 + 2, ticks + assert len(deletes) == 2, ticks + assert len(ticks) == len(selects) + len(deletes), ticks + # None of them asks, row by row, whether its schedule has a newer tick. + # That question is kept for a row whose schedule's kept tick has gone: + # on MySQL each answer walks the schedule's history up to the row, and + # a dry run, which deletes nothing on the way, would walk all of it for + # every row. + assert not [s for s in ticks if "EXISTS" in s], ticks + assert "Deleted 6 schedule tick row(s)" in out + assert OxScheduleTick.objects.count() == 3 + + +@pytest.mark.django_db(transaction=True) +def test_a_key_in_two_spellings_is_pruned_a_batch_at_a_time(): + # MySQL's collation takes "Report" and "report" for one key: one unique + # index entry a tick, one history to every read by key, and one of the + # two spellings back from the read of every schedule's key. The rows + # that hold the other spelling are still checked with their batch. + # Asked row by row instead, whether its schedule has a newer tick, each + # answer walks the schedule's history up to the row. + if connection.vendor != "mysql": + pytest.skip( + "MySQL compares schedule keys by a collation that ignores case; " + f"this run is on {connection.vendor}, which does not" + ) + for days in (40, 30, 20): + make_tick("Report", days_ago=days) + for days in (35, 25, 15): + make_tick("report", days_ago=days) + newest = make_tick("report", days_ago=1) + + with CaptureQueriesContext(connection) as queries: + out, _ = prune("--batch-size", "4") + + statements = [_sql(q["sql"]).lstrip() for q in queries.captured_queries] + ticks = [s for s in statements if TICKS.upper() in s] + assert not [s for s in ticks if "EXISTS" in s], ticks + assert "Deleted 6 schedule tick row(s)" in out + assert remaining() == {newest} + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("args", [(), ("--dry-run",)], ids=["run", "dry-run"]) +def test_a_schedule_first_dispatched_during_the_prune_keeps_its_tick(args): + # Its key was not among those read at the start, and its one tick is + # older than a cutoff of now: the row is its schedule's whole history. + make_tick("k", days_ago=30) + kept = make_tick("k", days_ago=10) + first = timezone.now() - timedelta(hours=1) + state = {"keys_read": False, "done": False} + + def wrapper(execute, sql, params, many, context): + # Before the statement that follows the read of the keys, when that + # read's cursor is closed and SQLite lets another connection write. + if state["keys_read"] and not state["done"]: + state["done"] = True + in_another_connection( + f"INSERT INTO {TICKS} " # noqa: S608 + "(schedule_name, scheduled_for, task_id, created_at) " + "VALUES ('late', %s, NULL, %s)", + [connection.ops.adapt_datetimefield_value(first)] * 2, + ) + result = execute(sql, params, many, context) + text = _sql(sql) + if "DISTINCT" in text and TICKS.upper() in text: + state["keys_read"] = True + return result + + before = remaining() + with connection.execute_wrapper(wrapper): + out, _ = prune("--older-than", "0", "--batch-size", "1", *args) + + assert state["done"], "the prune never read the schedule keys" + late = OxScheduleTick.objects.get(schedule_name="late").pk + assert remaining() == (before | {late} if args else {kept, late}) + verb, keeps = ("Would delete", "Would keep") if args else ("Deleted", "Kept") + assert f"{verb} 1 schedule tick row(s)" in out + assert f"{keeps} 2 schedule tick row(s) scheduled before " in out + + +@pytest.mark.django_db +class TestPurgeTakesATarget: + def test_the_flag_alone_is_refused(self): + bad = insert_tick("k", OLDEST_HERE[connection.vendor]) + + with pytest.raises(CommandError) as info: + prune("--purge-unreadable-ticks") + + assert str(info.value) == ( + "--purge-unreadable-ticks needs at least one --schedule-key or " + "--tick-pk. On MySQL, --schedule-key uses the tick table's collation, " + "which may match keys differing in case, accents or trailing spaces. " + "Inspect --dry-run before removing rows, or use --tick-pk to target " + "individual rows." + ) + assert remaining() == {bad} + + @pytest.mark.parametrize( + "args", [("--schedule-key", "k"), ("--tick-pk", "1")], ids=["key", "pk"] + ) + def test_a_target_without_the_flag_is_refused(self, args): + bad = insert_tick("k", OLDEST_HERE[connection.vendor]) + + with pytest.raises(CommandError) as info: + prune(*args) + + assert str(info.value) == ( + "Use --schedule-key and --tick-pk with --purge-unreadable-ticks." + ) + assert remaining() == {bad} + + def test_prune_options_beside_it_are_refused(self): + with pytest.raises(CommandError) as info: + prune( + "--purge-unreadable-ticks", + "--schedule-key", + "k", + "--queue", + "emails", + "--older-than", + "1d", + "--include-failed", + ) + + assert str(info.value) == ( + "--purge-unreadable-ticks cannot be combined with --queue, " + "--older-than, --include-failed. Run ordinary pruning separately." + ) + + +@pytest.fixture +def two_schedules(): + """ + Schedule "k" with an unreadable oldest and newest tick around readable + ones, and schedule "j" with one unreadable tick. Their keys by name. + """ + vendor = connection.vendor + return { + "k_oldest": insert_tick("k", OLDEST_HERE[vendor]), + "k_readable": [make_tick("k", days_ago=days) for days in (30, 1)], + "k_newest": insert_tick("k", NEWEST_HERE[vendor]), + "j": insert_tick("j", OLDEST_HERE[vendor]), + } + + +@pytest.mark.django_db +class TestPurgeUnreadableTicks: + def test_dry_run_names_each_tick_and_changes_nothing(self, two_schedules): + before = remaining() + + out, err = prune("--purge-unreadable-ticks", "--schedule-key", "k", "--dry-run") + + assert remaining() == before + lines = out.splitlines() + assert lines[0] == "Would remove 2 unreadable schedule tick row(s):" + for pk in (two_schedules["k_oldest"], two_schedules["k_newest"]): + assert any( + line.startswith(f" tick {pk}, schedule 'k': scheduled_for ") + for line in lines + ), out + assert err.startswith("Warning: removing dispatch history can make") + + def test_removes_only_the_named_schedules_unreadable_ticks(self, two_schedules): + out, err = prune("--purge-unreadable-ticks", "--schedule-key", "k") + + assert remaining() == {*two_schedules["k_readable"], two_schedules["j"]} + assert out.startswith("Removed 2 unreadable schedule tick row(s):") + assert "Warning: removing dispatch history" in err + + def test_json_names_what_it_removed(self, two_schedules): + out, _ = prune( + "--purge-unreadable-ticks", + "--schedule-key", + "k", + "--schedule-key", + "j", + "--format", + "json", + ) + + data = json.loads(out) + assert out.strip() == json.dumps(data) + assert data["tick_rows"] == 3 + assert data["dry_run"] is False + assert data["schedule_keys"] == ["k", "j"] + assert {named["pk"] for named in data["unreadable_ticks"]} == { + two_schedules["k_oldest"], + two_schedules["k_newest"], + two_schedules["j"], + } + assert data["kept_ticks"] == [] + assert data["missing_tick_pks"] == [] + assert remaining() == set(two_schedules["k_readable"]) + + def test_by_tick_pk_removes_only_that_tick(self, two_schedules): + readable = two_schedules["k_readable"][0] + + out, _ = prune( + "--purge-unreadable-ticks", + "--tick-pk", + str(two_schedules["j"]), + "--tick-pk", + str(readable), + "--tick-pk", + "987654321", + ) + + assert two_schedules["j"] not in remaining() + assert remaining() == { + two_schedules["k_oldest"], + two_schedules["k_newest"], + *two_schedules["k_readable"], + } + assert out.startswith("Removed 1 unreadable schedule tick row(s):") + assert f" tick {two_schedules['j']}, schedule 'j': " in out + assert ( + f"Kept tick {readable}, schedule 'k': the row is readable, with " + "scheduled_for " in out + ) + assert "No tick row 987654321." in out + + @pytest.mark.parametrize( + "number", + [2**63, -(2**63) - 1, 10**30], + ids=["one-past-the-largest", "one-before-the-smallest", "thirty-digits"], + ) + def test_a_number_no_primary_key_can_be_names_no_tick(self, two_schedules, number): + # Beside a tick that is there and does not read, which still goes. + out, _ = prune( + "--purge-unreadable-ticks", + "--tick-pk", + str(number), + "--tick-pk", + str(two_schedules["j"]), + ) + + assert two_schedules["j"] not in remaining() + assert len(remaining()) == 4 + assert out.startswith("Removed 1 unreadable schedule tick row(s):") + assert f"No tick row {number}." in out + + def test_a_schedule_with_nothing_unreadable_is_left_alone(self): + ticks = {make_tick("k", days_ago=days) for days in (30, 1)} + + out, err = prune("--purge-unreadable-ticks", "--schedule-key", "k") + + assert remaining() == ticks + assert out == "Removed 0 unreadable schedule tick row(s).\n" + assert err == "" + + def test_each_removed_tick_is_logged_by_event(self, two_schedules, caplog): + with caplog.at_level(logging.WARNING, logger="django_ox"): + prune("--purge-unreadable-ticks", "--tick-pk", str(two_schedules["j"])) + + (record,) = [ + r + for r in caplog.records + if getattr(r, "event", None) == "schedule_tick_history_removed" + ] + assert record.database == DEFAULT_DB_ALIAS + assert record.schedule_key == "j" + assert record.tick_pk == two_schedules["j"] + assert record.reason.startswith("scheduled_for ") + + def test_removes_without_loading_the_rows(self, two_schedules): + # With anything listening for delete signals, Django's delete() loads + # each row first, and Django's read of this one raises, or on MySQL + # with USE_TZ off reads the zero date as text. + try: + with transaction.atomic(): + loaded = OxScheduleTick.objects.get(pk=two_schedules["j"]) + except Exception: # what Django's read raised; nothing to keep + loaded = None + assert loaded is None or not isinstance(loaded.scheduled_for, datetime) + + def listener(sender, **kwargs): + pass + + pre_delete.connect(listener, weak=False, dispatch_uid="test_purge_listener") + try: + out, _ = prune("--purge-unreadable-ticks", "--schedule-key", "j") + finally: + pre_delete.disconnect(dispatch_uid="test_purge_listener") + + assert two_schedules["j"] not in remaining() + assert out.startswith("Removed 1 unreadable schedule tick row(s):") + + def test_a_failure_part_way_reports_what_is_already_gone( + self, two_schedules, caplog + ): + out = StringIO() + with ( + caplog.at_level(logging.WARNING, logger="django_ox"), + failing( + "DELETE FROM DJANGO_OX_OXSCHEDULETICK", + lambda: simulated("mysql-deadlock"), + lambda n: n == 2, + ), + pytest.raises(DatabaseError), + ): + call_command( + "ox_prune", + "--purge-unreadable-ticks", + "--schedule-key", + "k", + "--batch-size", + "1", + "--format", + "json", + stdout=out, + stderr=StringIO(), + ) + + data = json.loads(out.getvalue()) + assert data["tick_rows"] == 1 + ((gone,),) = [data["unreadable_ticks"]] + assert gone["pk"] not in remaining() + assert ( + len(remaining() & {two_schedules["k_oldest"], two_schedules["k_newest"]}) + == 1 + ) + assert [ + r.tick_pk + for r in caplog.records + if getattr(r, "event", None) == "schedule_tick_history_removed" + ] == [gone["pk"]] + + def test_a_tick_whose_schedule_key_cannot_be_handed_over_goes_by_its_pk(self): + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + readable = {make_tick("k", days_ago=days) for days in (30, 1)} + bad = insert_tick_keyed(UNDECODABLE_KEY, days_ago=30) + + out, _ = prune("--purge-unreadable-ticks", "--tick-pk", str(bad)) + + assert remaining() == readable + assert out.startswith("Removed 1 unreadable schedule tick row(s):") + assert f" tick {bad}, schedule '': schedule_name could not be read" in out + + @pytest.mark.parametrize("fmt", ["text", "json"]) + def test_a_tick_whose_schedule_key_is_not_text_goes_by_its_primary_key(self, fmt): + only_on("sqlite", "SQLite keeps bytes where a schedule key should be") + bad = insert_tick_keyed(BYTES_KEY, days_ago=30) + other = insert_tick_keyed(BYTES_KEY, days_ago=1) + readable = make_tick("k", days_ago=30) + + out, _ = prune( + "--purge-unreadable-ticks", "--tick-pk", str(bad), "--format", fmt + ) + + assert remaining() == {other, readable} + reason = "schedule_name holds b'nightly', which is not text" + if fmt == "json": + data = json.loads(out) + assert data["unreadable_ticks"] == [ + {"pk": bad, "schedule_name": "", "reason": reason} + ] + else: + assert out.startswith("Removed 1 unreadable schedule tick row(s):") + assert f" tick {bad}, schedule '': {reason}" in out + + def test_a_tick_the_driver_cannot_hand_over_is_removed(self): + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + readable = [make_tick("k", days_ago=days) for days in (30, 20, 10, 1)] + bad = insert_tick("k", "CAST(X'30303030FF' AS TEXT)") + + out, _ = prune( + "--purge-unreadable-ticks", "--schedule-key", "k", "--batch-size", "2" + ) + + assert remaining() == set(readable) + assert f" tick {bad}, schedule 'k': scheduled_for could not be read" in out + + +@pytest.mark.django_db(transaction=True) +def test_a_tick_repaired_since_it_was_found_is_not_removed(): + # Between the read that found the tick and its removal, the tick is + # repaired on another connection. The removal reads each row again + # under its lock, and only rows that still cannot be read go. + bad = insert_tick("k", OLDEST_HERE[connection.vendor]) + repaired_to = (timezone.now() - timedelta(days=3)).replace(microsecond=0) + state = {"done": False} + + def wrapper(execute, sql, params, many, context): + text = _sql(sql).lstrip() + removing = TICKS.upper() in text and ( + "FOR UPDATE" in text or text.startswith("UPDATE") + ) + if removing and not state["done"]: + state["done"] = True + in_another_connection( + f"UPDATE {TICKS} SET scheduled_for = %s WHERE id = %s", # noqa: S608 + [connection.ops.adapt_datetimefield_value(repaired_to), bad], + ) + return execute(sql, params, many, context) + + with connection.execute_wrapper(wrapper): + out, _ = prune("--purge-unreadable-ticks", "--schedule-key", "k") + + assert state["done"], "the purge never reached its removal" + assert remaining() == {bad} + assert OxScheduleTick.objects.get(pk=bad).scheduled_for == repaired_to + assert out.startswith("Removed 0 unreadable schedule tick row(s).") + assert ( + f"Kept tick {bad}, schedule 'k': the row is now readable, with scheduled_for " + in out + ) + + +# -- how a tick row is deleted ------------------------------------------------ + + +class Listening: + """ + Receivers for both delete signals of every model, connected for a + block. `sent` is each one sent, as (signal, model, primary key). + """ + + def __init__(self): + self.sent = [] + + def _pre(self, sender, instance, **kwargs): + self.sent.append(("pre_delete", sender, instance.pk)) + + def _post(self, sender, instance, **kwargs): + self.sent.append(("post_delete", sender, instance.pk)) + + def __enter__(self): + pre_delete.connect(self._pre, weak=False, dispatch_uid="test_prune_pre") + post_delete.connect(self._post, weak=False, dispatch_uid="test_prune_post") + return self + + def __exit__(self, *exc_info): + pre_delete.disconnect(dispatch_uid="test_prune_pre") + post_delete.disconnect(dispatch_uid="test_prune_post") + + def of(self, model): + return [(signal, pk) for signal, sender, pk in self.sent if sender is model] + + +class Deletes: + """connection.execute_wrapper() keeping each DELETE on the tick table as sent.""" + + def __init__(self): + self.sent = [] + + def __call__(self, execute, sql, params, many, context): + if _sql(sql).lstrip().startswith("DELETE FROM DJANGO_OX_OXSCHEDULETICK"): + self.sent.append((sql, list(params))) + return execute(sql, params, many, context) + + +def a_finished_task(*, days_ago): + when = timezone.now() - timedelta(days=days_ago) + return OxTask.objects.create( + task_path="tests.tasks.add", + backend_name="default", + queue_name="default", + enqueued_at=when, + status=OxTask.Status.SUCCESSFUL, + finished_at=when, + ) + + +#: Each way the command removes tick rows, as its arguments. +REMOVALS = [ + pytest.param((), id="prune"), + pytest.param(("--purge-unreadable-ticks", "--schedule-key", "k"), id="purge-key"), + pytest.param(("--purge-unreadable-ticks", "--tick-pk", "{oldest}"), id="purge-pk"), +] + + +def some_of_each(): + """ + Schedule "k" with an unreadable oldest and newest tick around three + that read, two of them past the cutoff, and schedule "h" with only + ticks that read. Their keys by name. + """ + vendor = connection.vendor + return { + "oldest": insert_tick("k", OLDEST_HERE[vendor]), + "k_old": [make_tick("k", days_ago=days) for days in (40, 30)], + "k_kept": make_tick("k", days_ago=1), + "newest": insert_tick("k", NEWEST_HERE[vendor]), + "h_old": [make_tick("h", days_ago=days) for days in (40, 30)], + "h_kept": make_tick("h", days_ago=1), + } + + +def gone_by(args, ticks): + """The tick rows a run with these arguments removes, of `some_of_each`.""" + if not args: + return {*ticks["k_old"], *ticks["h_old"]} + if "--tick-pk" in args: + return {ticks["oldest"]} + return {ticks["oldest"], ticks["newest"]} + + +@pytest.mark.django_db +class TestHowATickRowIsDeleted: + """ + A tick row goes in a DELETE of the command's own, by primary key: the + same statement for a batch of ticks that read as for one holding a + tick that does not, in ordinary pruning and in a named removal. No + delete signal is sent for a tick row, since sending one means loading + the row. Task rows are deleted as they were, and send theirs. + """ + + @pytest.mark.parametrize("args", REMOVALS) + def test_no_delete_signal_is_sent_for_a_tick_row(self, args): + ticks = some_of_each() + before = remaining() + args = [arg.format(oldest=ticks["oldest"]) for arg in args] + + with Listening() as listening: + prune(*args) + + assert before - remaining() == gone_by(args, ticks) + assert listening.of(OxScheduleTick) == [] + + def test_nor_for_a_batch_in_which_every_tick_reads(self): + old = [make_tick("h", days_ago=days) for days in (40, 30)] + kept = make_tick("h", days_ago=1) + + with Listening() as listening: + out, err = prune() + + assert remaining() == {kept} + assert "Deleted 2 schedule tick row(s)" in out + assert err == "" + assert set(old).isdisjoint(remaining()) + assert listening.sent == [] + + def test_a_task_row_still_sends_both_of_its_own(self): + old = [a_finished_task(days_ago=30) for _ in range(3)] + recent = a_finished_task(days_ago=1) + when = timezone.now() - timedelta(days=30) + # A tick that points at a task that goes: kept as its schedule's + # newest, with its task cleared and no signal of its own. + tick = OxScheduleTick.objects.create( + schedule_name="h", scheduled_for=when, task=old[0], created_at=when + ) + + with Listening() as listening: + out, _ = prune() + + assert "Deleted 3 SUCCESSFUL/DISCARDED task row(s)" in out + assert set(OxTask.objects.values_list("pk", flat=True)) == {recent.pk} + sent = listening.of(OxTask) + assert sorted(pk for signal, pk in sent if signal == "pre_delete") == sorted( + task.pk for task in old + ) + assert sorted(pk for signal, pk in sent if signal == "post_delete") == sorted( + task.pk for task in old + ) + assert len(sent) == 6 + assert OxScheduleTick.objects.get(pk=tick.pk).task_id is None + assert listening.of(OxScheduleTick) == [] + + @pytest.mark.parametrize("args", REMOVALS) + def test_it_goes_in_one_statement_of_the_commands_own(self, args): + ticks = some_of_each() + args = [arg.format(oldest=ticks["oldest"]) for arg in args] + expected = sorted(gone_by(args, ticks)) + deletes = Deletes() + + with connection.execute_wrapper(deletes): + prune(*args) + + # The table and its key quoted as this database quotes them, and + # the keys as parameters: none of them written into the statement. + quote = connection.ops.quote_name + table, key = quote(TICKS), quote("id") + # One statement for the batch: every old row ordinary pruning read, + # or every row a named removal found. + ((statement, params),) = deletes.sent + marks = ", ".join(["%s"] * len(params)) + assert statement == " ".join( + ["DELETE FROM", table, "WHERE", key, f"IN ({marks})"] + ) + assert not [char for char in statement if char.isdigit()] + assert all(type(pk) is int for pk in params) + assert sorted(params) == expected + + @pytest.mark.parametrize("args", REMOVALS[:2]) + def test_a_statement_carries_no_more_keys_than_a_batch(self, args): + ticks = some_of_each() + deletes = Deletes() + + with connection.execute_wrapper(deletes): + prune(*args, "--batch-size", "1") + + removed = gone_by(args, ticks) + assert [len(params) for _, params in deletes.sent] == [1] * len(removed) + assert {pk for _, (pk,) in deletes.sent} == removed + + @pytest.mark.parametrize("args", REMOVALS) + def test_no_tick_row_goes_through_a_delete_of_djangos(self, args, monkeypatch): + ticks = some_of_each() + before = remaining() + args = [arg.format(oldest=ticks["oldest"]) for arg in args] + raw_delete = QuerySet._raw_delete + through_django = [] + + def recording(queryset, using): + through_django.append(queryset.model) + return raw_delete(queryset, using) + + monkeypatch.setattr(QuerySet, "_raw_delete", recording) + with CaptureQueriesContext(connection) as queries: + prune(*args) + + assert before - remaining() == gone_by(args, ticks) + assert OxScheduleTick not in through_django + # Nor is a tick row read to be deleted: no statement takes the + # column a delete of Django's would load every row for. + read_whole = [ + q["sql"] + for q in queries.captured_queries + if _sql(q["sql"]).lstrip().startswith("SELECT") + and f"{TICKS.upper()}.CREATED_AT" in _sql(q["sql"]) + ] + assert read_whole == [] + + def test_a_dry_run_sends_none(self): + some_of_each() + before = remaining() + deletes = Deletes() + + with Listening() as listening, connection.execute_wrapper(deletes): + prune("--dry-run") + prune("--purge-unreadable-ticks", "--schedule-key", "k", "--dry-run") + + assert remaining() == before + assert deletes.sent == [] + assert listening.sent == [] + + +@pytest.mark.django_db(transaction=True) +def test_a_named_removal_deletes_inside_the_transaction_that_read_the_rows_again(): + # With no transaction of the caller's, so the command's own shows: the + # batch is locked and read again, and its DELETE goes before the + # commit that lets the locks go. + bad = insert_tick("k", OLDEST_HERE[connection.vendor]) + + with CaptureQueriesContext(connection) as queries: + prune("--purge-unreadable-ticks", "--tick-pk", str(bad)) + + words = [_sql(q["sql"]).split()[0] for q in queries.captured_queries] + removal = words[words.index("BEGIN") :] + locking = ( + ["SELECT"] + if connection.features.has_select_for_update + else ["UPDATE", "SELECT"] + ) + assert removal == ["BEGIN", *locking, "DELETE", "COMMIT"] + assert remaining() == set() + + +# -- a settings schedule an unreadable tick holds up ------------------------- + +BLOCKED = "blocked-every-second" +CONTROL = "control-every-second" + + +def command(root, *args): + """A real `manage.py` run in the project at `root`: (exit code, out, err).""" + env = dict(os.environ) + env["DJANGO_SETTINGS_MODULE"] = "isoproj.settings" + env["OX_TEST_DB_NAME"] = str(connection.settings_dict["NAME"]) + # The arguments are this test's own literals. + done = subprocess.run( # noqa: S603 + [sys.executable, "manage.py", *args], + cwd=root, + env=env, + capture_output=True, + text=True, + timeout=GUARD, + check=False, + ) + return done.returncode, done.stdout, done.stderr + + +def ran(label): + """How many tasks a schedule labelled `label` has enqueued that ran.""" + finished = OxTask.objects.filter(status=OxTask.Status.SUCCESSFUL) + return sum(args == [label] for args in finished.values_list("args", flat=True)) + + +def run_worker_until(root, log, what, predicate): + proc = start(root, log, batch=False) + try: + if not wait_for(predicate, timeout=GUARD): + pytest.fail(f"no {what} within {GUARD}s:\n{log.read_text()[-4000:]}") + finally: + if proc.poll() is None: + proc.send_signal(signal.SIGTERM) + finish(proc, log) + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.skipif(os.name != "posix", reason="worker processes") +def test_a_settings_schedule_held_up_by_an_unreadable_tick_fires_after_its_removal( + tmp_path, +): + every_second = {"task": "tests.tasks.labelled", "every": 1} + root = project( + tmp_path, + { + BLOCKED: {**every_second, "args": [BLOCKED]}, + CONTROL: {**every_second, "args": [CONTROL]}, + }, + ) + bad = insert_tick(BLOCKED, OLDEST_HERE[connection.vendor]) + + # Held up: the schedule beside it fires twice, and this one never does. + run_worker_until( + root, tmp_path / "first.log", "second control run", lambda: ran(CONTROL) >= 2 + ) + assert ran(BLOCKED) == 0 + blocked = OxScheduleTick.objects.filter(schedule_name=BLOCKED) + assert list(blocked.values_list("pk", flat=True)) == [bad] + + # Ordinary pruning keeps it, completes and says so. + code, out, err = command(root, "ox_prune", "--format", "json") + assert code == 0, err + data = json.loads(out) + assert [named["pk"] for named in data["unreadable_ticks"]] == [bad] + assert f" tick {bad}, schedule '{BLOCKED}': " in err + + # A dry run names it and leaves it. + code, out, err = command( + root, + "ox_prune", + "--purge-unreadable-ticks", + "--schedule-key", + BLOCKED, + "--dry-run", + ) + assert code == 0, err + assert out.startswith("Would remove 1 unreadable schedule tick row(s):") + assert OxScheduleTick.objects.filter(pk=bad).exists() + + code, out, err = command( + root, "ox_prune", "--purge-unreadable-ticks", "--schedule-key", BLOCKED + ) + assert code == 0, err + assert out.startswith("Removed 1 unreadable schedule tick row(s):") + assert "Warning: removing dispatch history" in err + assert not OxScheduleTick.objects.filter(schedule_name=BLOCKED).exists() + + # Started again, the schedule is a first sighting: it anchors without + # running, then fires from its next tick on. + run_worker_until( + root, + tmp_path / "second.log", + "run of the released schedule", + lambda: ran(BLOCKED) >= 1, + ) + history = list( + OxScheduleTick.objects.filter(schedule_name=BLOCKED) + .order_by("scheduled_for") + .values_list("scheduled_for", "task_id") + ) + assert history[0][1] is None, history + assert all(task is not None for _, task in history[1:]), history + assert len(history) >= 2, history + + +#: Four spellings of one key and, for each, an unreadable newest tick. They +#: differ in case, in an accent and in a trailing space, which a database +#: compares as equal or not by the collation of the column. +SPELLINGS = ["nightly", "Nightly", "nightly ", "nightl\u00fd"] + + +def spelled_ticks(): + """An unreadable tick under each spelling, with a readable one beside it.""" + literals = { + "postgresql": lambda i: f"'{10000 + i}-01-01 00:00:00+00'", + "mysql": lambda i: f"'9999-00-0{i + 1} 00:00:00'", + "sqlite": lambda i: f"'banana {i}'", + }[connection.vendor] + unreadable = { + key: insert_tick(key, literals(index)) for index, key in enumerate(SPELLINGS) + } + readable = { + key: make_tick(key, days_ago=3 + index) for index, key in enumerate(SPELLINGS) + } + return unreadable, readable + + +def key_collation(): + """The collation MySQL compares the tick table's schedule keys by.""" + with connection.cursor() as cursor: + cursor.execute( + "SELECT COLLATION_NAME FROM information_schema.COLUMNS " + "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s " + "AND COLUMN_NAME = 'schedule_name'", + [TICKS], + ) + return cursor.fetchone()[0] + + +def spellings_the_database_calls_nightly(): + """ + The spellings above that `schedule_name = 'nightly'` matches here: only + the exact one where keys compare exactly (PostgreSQL, SQLite), and on + MySQL the ones its collation calls equal. + """ + if connection.vendor != "mysql": + return {"nightly"} + collation = key_collation() + if collation == "utf8mb4_unicode_ci": + # Ignores case and accents, and pads with spaces when comparing. + return set(SPELLINGS) + if collation == "utf8mb4_0900_ai_ci": + # Ignores case and accents, and does not pad. + return set(SPELLINGS) - {"nightly "} + pytest.skip(f"no expectation is written for the collation {collation}") + + +@pytest.mark.django_db +class TestAKeyIsWhatTheDatabaseCallsTheSameKey: + """ + `--schedule-key` selects with the database's own comparison of the key. + PostgreSQL and SQLite compare exactly. MySQL compares by the collation + of the column, so a purge by one spelling can take the unreadable ticks + of spellings it calls equal. The dry run and the purge agree with each + other either way, and a tick named by its primary key is that tick alone. + """ + + def test_a_purge_takes_the_spellings_the_comparison_matches(self): + unreadable, readable = spelled_ticks() + expected = {unreadable[key] for key in spellings_the_database_calls_nightly()} + + out, _ = prune( + "--purge-unreadable-ticks", "--schedule-key", "nightly", "--format", "json" + ) + + data = json.loads(out) + assert {named["pk"] for named in data["unreadable_ticks"]} == expected + assert remaining() == ( + set(unreadable.values()) - expected | set(readable.values()) + ) + + def test_the_dry_run_names_what_the_purge_then_removes(self): + unreadable, _ = spelled_ticks() + before = remaining() + + out, _ = prune( + "--purge-unreadable-ticks", + "--schedule-key", + "nightly", + "--dry-run", + "--format", + "json", + ) + listed = {named["pk"] for named in json.loads(out)["unreadable_ticks"]} + assert remaining() == before + + prune("--purge-unreadable-ticks", "--schedule-key", "nightly") + + assert before - remaining() == listed + assert listed == { + unreadable[key] for key in spellings_the_database_calls_nightly() + } + + def test_a_readable_tick_is_never_taken(self): + _, readable = spelled_ticks() + + prune("--purge-unreadable-ticks", "--schedule-key", "nightly") + + assert set(readable.values()) <= remaining() + + @pytest.mark.parametrize("spelling", SPELLINGS) + def test_a_tick_pk_removes_that_tick_whatever_the_comparison_says(self, spelling): + unreadable, readable = spelled_ticks() + + out, _ = prune( + "--purge-unreadable-ticks", "--tick-pk", str(unreadable[spelling]) + ) + + assert out.startswith("Removed 1 unreadable schedule tick row(s):") + assert remaining() == ( + set(unreadable.values()) - {unreadable[spelling]} | set(readable.values()) + ) diff --git a/tests/test_replica_reads.py b/tests/test_replica_reads.py index 1638e93..40f51e0 100644 --- a/tests/test_replica_reads.py +++ b/tests/test_replica_reads.py @@ -38,7 +38,7 @@ from django.utils import timezone from django_ox import actions, metrics, stats -from django_ox.models import OxTask +from django_ox.models import OxScheduleTick, OxTask # The replica is `default` and the primary is `alt`, so a statement that # routes itself lands somewhere a pinned one does not. Both aliases are @@ -187,6 +187,65 @@ def test_prune_does_not_loop_over_a_replica_that_never_empties(self): call_command("ox_prune", "--older-than", "1d", "--batch-size", "3") assert OxTask.objects.using(PRIMARY).count() == 0 + def _old_ticks(self): + """ + Two old ticks and a recent one of one schedule on the primary, and + on the replica a snapshot from before the recent one. Their keys. + """ + now = timezone.now() + primary = OxScheduleTick.objects.using(PRIMARY) + pks = [] + for days in (40, 30, 1): + when = now - timedelta(days=days) + tick = primary.create( + schedule_name="k", scheduled_for=when, created_at=when + ) + pks.append(tick.pk) + if days != 1: + OxScheduleTick.objects.using(REPLICA).create( + pk=tick.pk, schedule_name="k", scheduled_for=when, created_at=when + ) + return pks + + def test_prune_deletes_tick_rows_on_the_primary(self): + first, second, newest = self._old_ticks() + out = StringIO() + with no_statement_on(REPLICA): + call_command("ox_prune", "--older-than", "7d", stdout=out) + assert "Deleted 2 schedule tick row(s)" in out.getvalue() + primary = OxScheduleTick.objects.using(PRIMARY) + assert list(primary.values_list("pk", flat=True)) == [newest] + # Nothing was deleted on the replica: it is still the snapshot. + replica = OxScheduleTick.objects.using(REPLICA) + assert sorted(replica.values_list("pk", flat=True)) == [first, second] + + def test_a_named_removal_deletes_tick_rows_on_the_primary(self): + first, second, newest = self._old_ticks() + # The primary is SQLite whatever the suite runs on, and keeps text + # that is not a date. The replica is given a row under the same key + # that reads, which a removal sent there would leave in place. + with connections[PRIMARY].cursor() as cursor: + cursor.execute( + "UPDATE django_ox_oxscheduletick SET scheduled_for = 'banana' " + "WHERE id = %s", + [first], + ) + out = StringIO() + with no_statement_on(REPLICA): + call_command( + "ox_prune", + "--purge-unreadable-ticks", + "--tick-pk", + str(first), + stdout=out, + stderr=StringIO(), + ) + assert out.getvalue().startswith("Removed 1 unreadable schedule tick row(s):") + primary = OxScheduleTick.objects.using(PRIMARY) + assert sorted(primary.values_list("pk", flat=True)) == [second, newest] + replica = OxScheduleTick.objects.using(REPLICA) + assert sorted(replica.values_list("pk", flat=True)) == [first, second] + def test_health_reports_the_primary_backlog(self): self._backlog(40) with no_statement_on(REPLICA), pytest.raises(CommandError) as caught: diff --git a/tests/test_stored_read.py b/tests/test_stored_read.py index 5779149..5cc1fcd 100644 --- a/tests/test_stored_read.py +++ b/tests/test_stored_read.py @@ -53,6 +53,7 @@ from django_ox import _stored_read from django_ox._stored_read import ( TickRead, + TickRow, UnreadableRow, UnreadableValue, as_text, @@ -66,6 +67,7 @@ lock_schedule, newest_tick_pk, read_schedules, + read_ticks, ) from django_ox.exceptions import StoredValueUnreadable from django_ox.models import OxSchedule, OxScheduleTick @@ -1488,15 +1490,17 @@ def the_reads(): keys = ["k0", "k1", "k2"] for key in keys: a_tick(key, now) + every_tick = OxScheduleTick.objects.order_by("pk") return { "rows": (SCHEDULES, lambda: read_schedules([r.pk for r in rows], using=ALIAS)), "newest": (TICKS, lambda: latest_ticks(keys, since, using=ALIAS)), "earliest": (TICKS, lambda: earliest_tick("k0", using=ALIAS)), + "ticks": (TICKS, lambda: read_ticks(every_tick, using=ALIAS)), } @pytest.mark.django_db -@pytest.mark.parametrize("read", ["rows", "newest", "earliest"]) +@pytest.mark.parametrize("read", ["rows", "newest", "earliest", "ticks"]) def test_in_a_transaction_a_read_the_database_refuses_is_raised_as_it_was(read): """ A statement the server refuses ends a PostgreSQL transaction for every @@ -1576,8 +1580,30 @@ def test_in_a_transaction_a_refused_read_of_one_column_is_raised(): refusing.assert_raised_as_it_was(caught) +@pytest.mark.django_db +def test_in_a_transaction_a_refused_read_of_the_tick_rows_keys_is_raised(): + """ + Tick rows the driver could not hand over are read again as their keys + and then by key, and each of those reads is held to the same rule. + """ + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + a_tick("k", now_ish()) + insert_tick("bad", NOT_UTF_8) + # The rows whole, which the driver fails, and then their keys. + refusing = Refusing(TICKS, nth=2) + + with ( + pytest.raises(DataError) as caught, + transaction.atomic(), + connection.execute_wrapper(refusing), + ): + read_ticks(OxScheduleTick.objects.order_by("pk"), using=ALIAS) + + refusing.assert_raised_as_it_was(caught) + + @pytest.mark.django_db(transaction=True) -@pytest.mark.parametrize("read", ["rows", "newest", "earliest"]) +@pytest.mark.parametrize("read", ["rows", "newest", "earliest", "ticks"]) def test_with_no_transaction_open_a_read_the_database_refuses_is_raised_as_it_was(read): """ With no transaction open the refused statement was a transaction of its @@ -1697,6 +1723,37 @@ def test_with_no_transaction_open_a_read_that_fails_on_one_key_is_narrowed_to_it assert answers["k2"] == TickRead("k2", at=now) +@pytest.mark.django_db(transaction=True) +def test_with_no_transaction_open_a_read_that_fails_on_one_tick_row_is_narrowed_to_it(): + """ + And for tick rows: the selection is read again as its keys, which + always read, and then by key in halves until the row that fails is + alone. That row is answered as one whose tick does not read, under the + schedule key it carries, and every other row as it stands. + """ + now = now_ish() + ticks = [a_tick(f"k{i}", now - timedelta(minutes=i)) for i in range(4)] + bad = ticks[2] + failing = Failing(TICKS, "scheduled_for", bad.pk) + selection = OxScheduleTick.objects.filter(pk__in=[t.pk for t in ticks]) + + assert connection.get_autocommit() + assert not connection.in_atomic_block + with connection.execute_wrapper(failing): + rows = read_ticks(selection.order_by("pk"), using=ALIAS) + + # The selection, its rows by key, the half the row is in, the row alone. + assert len(failing.failed) == 4 + assert [row.pk for row in rows] == [tick.pk for tick in ticks] + assert rows[2].unreadable + assert rows[2].key == "k2" + assert rows[2].reason.startswith("scheduled_for could not be read: ") + assert type(rows[2].cause) is Failing.RAISES[connection.vendor] + for index in (0, 1, 3): + tick = ticks[index] + assert rows[index] == TickRow(tick.pk, f"k{index}", at=tick.scheduled_for) + + @pytest.mark.django_db(transaction=True) def test_with_no_transaction_open_text_the_driver_cannot_decode_is_narrowed_down(): """ @@ -2678,3 +2735,163 @@ def test_a_tick_row(self, vendor, literal): assert oldest.unreadable assert oldest.reason.startswith("scheduled_for holds ") assert earliest_tick("healthy", using=ALIAS).at == now + + +@pytest.mark.django_db +class TestReadingTickRows: + def test_reads_each_row_in_the_selections_order_in_one_statement(self): + now = now_ish() + made = [a_tick(f"k{i % 2}", now - timedelta(minutes=i)) for i in range(5)] + selection = OxScheduleTick.objects.order_by("scheduled_for") + + with CaptureQueriesContext(connection) as queries: + rows = read_ticks(selection, using=ALIAS) + + assert len(selects(queries, TICKS)) == 1 + assert [row.pk for row in rows] == [tick.pk for tick in reversed(made)] + for row in rows: + django = OxScheduleTick.objects.get(pk=row.pk) + assert row.key == django.schedule_name + assert same(row.at, django.scheduled_for) + assert not row.unreadable + assert read_ticks(selection, using=ALIAS, limit=2) == rows[:2] + assert read_ticks(selection[3:], using=ALIAS, limit=10) == rows[3:] + + @pytest.mark.parametrize( + ("vendor", "literal"), UNREADABLE_NEWEST_TICKS + UNREADABLE_OLDEST_TICKS + ) + def test_an_unreadable_tick_is_its_rows_answer_alone(self, vendor, literal): + only_on(vendor, "kept by that engine only") + now = now_ish() + before = a_tick("k", now - timedelta(hours=1)) + bad = insert_tick("k", literal) + after = a_tick("other", now) + assert misread(django_earliest("k")) or misread(django_newest("k")) + + rows = read_ticks(OxScheduleTick.objects.order_by("pk"), using=ALIAS) + + assert [row.pk for row in rows] == [before.pk, bad, after.pk] + assert rows[0] == TickRow(before.pk, "k", at=before.scheduled_for) + assert rows[2] == TickRow(after.pk, "other", at=now) + assert rows[1].unreadable + assert rows[1].key == "k" + assert rows[1].at is None + assert rows[1].reason.startswith("scheduled_for holds ") + assert isinstance(rows[1].cause, UnreadableValue) + + @pytest.mark.parametrize("position", ["first", "middle", "last"]) + def test_a_tick_the_driver_cannot_hand_over_is_found_by_halving(self, position): + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + now = now_ish() + index = {"first": 0, "middle": 3, "last": 7}[position] + pks = [] + for i in range(8): + if i == index: + pks.append(insert_tick("bad", NOT_UTF_8)) + else: + pks.append(a_tick(f"k{i}", now - timedelta(minutes=i)).pk) + + # In this test's transaction, and in no savepoint. + assert connection.in_atomic_block + with CaptureQueriesContext(connection) as queries: + rows = read_ticks(OxScheduleTick.objects.order_by("pk"), using=ALIAS) + + assert not savepoints(queries) + assert [row.pk for row in rows] == pks + for i, row in enumerate(rows): + if i == index: + assert row.unreadable + assert row.key == "bad" + assert "UTF-8" in row.reason + assert row.reason.startswith("scheduled_for could not be read") + else: + assert row == TickRow(pks[i], f"k{i}", at=now - timedelta(minutes=i)) + + def test_a_key_the_driver_cannot_hand_over_is_its_rows_answer_alone(self): + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + now = now_ish() + before = a_tick("k", now - timedelta(hours=1)) + by_sql( + f"INSERT INTO {TICKS} (schedule_name, scheduled_for, task_id, created_at) " # noqa: S608 + "VALUES (CAST(X'6B80FF' AS TEXT), '2026-10-05 00:00:00', NULL, " + "'2026-10-05 00:00:00')" + ) + after = a_tick("other", now) + (bad,) = set(OxScheduleTick.objects.values_list("pk", flat=True)) - { + before.pk, + after.pk, + } + + rows = read_ticks(OxScheduleTick.objects.order_by("pk"), using=ALIAS) + + assert [row.pk for row in rows] == [before.pk, bad, after.pk] + assert rows[0] == TickRow(before.pk, "k", at=before.scheduled_for) + assert rows[2] == TickRow(after.pk, "other", at=now) + assert rows[1].unreadable + assert rows[1].key == "" + assert rows[1].at is None + assert rows[1].reason.startswith("schedule_name could not be read") + assert "UTF-8" in rows[1].reason + + @pytest.mark.parametrize( + "tick", + ["'2026-10-05 00:00:00'", "'banana'", NOT_UTF_8], + ids=["tick-reads", "tick-does-not", "tick-not-utf-8"], + ) + def test_a_key_that_is_bytes_is_its_rows_answer_alone(self, tick): + # The driver hands bytes over as they are, so nothing raises: the + # row is answered as one whose key does not read, whatever its tick. + only_on("sqlite", "SQLite keeps bytes where a schedule key should be") + now = now_ish() + before = a_tick("k", now - timedelta(hours=1)) + by_sql( + f"INSERT INTO {TICKS} (schedule_name, scheduled_for, task_id, created_at) " # noqa: S608 + f"VALUES (X'6E696768746C79', {tick}, NULL, '2026-10-05 00:00:00')" + ) + after = a_tick("other", now) + (bad,) = set(OxScheduleTick.objects.values_list("pk", flat=True)) - { + before.pk, + after.pk, + } + + rows = read_ticks(OxScheduleTick.objects.order_by("pk"), using=ALIAS) + + assert [row.pk for row in rows] == [before.pk, bad, after.pk] + assert rows[0] == TickRow(before.pk, "k", at=before.scheduled_for) + assert rows[2] == TickRow(after.pk, "other", at=now) + assert rows[1] == TickRow( + bad, "", reason="schedule_name holds b'nightly', which is not text" + ) + + def test_a_column_the_read_does_not_take_is_not_its_concern(self): + only_on("postgresql", "PostgreSQL keeps year 10000") + now = now_ish() + tick = a_tick("k", now) + by_sql( + f"UPDATE {TICKS} SET created_at = '10000-01-01 00:00:00+00' WHERE id = %s", # noqa: S608 + [tick.pk], + ) + + assert read_ticks(OxScheduleTick.objects.all(), using=ALIAS) == [ + TickRow(tick.pk, "k", at=now) + ] + + def test_inside_a_transaction_the_transaction_still_works_after(self): + now = now_ish() + a_tick("k0", now) + literal = { + "postgresql": "'infinity'", + "mysql": "'9999-00-00 00:00:00'", + "sqlite": NOT_UTF_8, + }[connection.vendor] + insert_tick("k1", literal) + + with transaction.atomic(): + with CaptureQueriesContext(connection) as queries: + rows = read_ticks(OxScheduleTick.objects.order_by("pk"), using=ALIAS) + assert ( + OxScheduleTick.objects.filter(schedule_name="k0").update(task=None) == 1 + ) + + assert not savepoints(queries) + assert [row.unreadable for row in rows] == [False, True]