From ce6b7fc2c82ac9652ad19613311d6b27beb2bf25 Mon Sep 17 00:00:00 2001 From: Ricardo Gonzalez Date: Tue, 11 Aug 2026 19:06:19 -0700 Subject: [PATCH] Store one cache record per declared job The single jobs document coupled every job's fate: one lost read-merge-write race could clobber unrelated progress, eviction was all-or-nothing, and the cache's per-item size limit bounded the population. Declared-only made the document unnecessary: code is the index, so each declared job now has its own record and reads enumerate the declared ids. Reconciliation becomes read-repair. A record that is missing, unreadable, or fingerprinted for a different declared schedule is rebuilt from its declaration at the point of use, owner-fenced, with a log line that makes eviction observable. The takeover sweep, its convergence marker, and the quarantine machinery all delete; takeover syncs lazily because every read validates against code. --- .../per-job-cache-records.feature.md | 9 + integrations/vercel-apscheduler/README.md | 9 +- integrations/vercel-apscheduler/SCHEDULER.md | 75 +-- .../tests/unit/test_apscheduler_cache.py | 89 +--- .../unit/test_apscheduler_integration.py | 41 +- .../integrations/apscheduler/_adapter.py | 177 +------ .../apscheduler/_backends/_protocols.py | 15 +- .../apscheduler/_backends/cache/__init__.py | 10 +- .../apscheduler/_backends/cache/_doc.py | 12 +- .../apscheduler/_backends/cache/_driver.py | 23 - .../apscheduler/_backends/cache/_jobstore.py | 444 +++++++++--------- 11 files changed, 350 insertions(+), 554 deletions(-) create mode 100644 changes/vercel-apscheduler/per-job-cache-records.feature.md diff --git a/changes/vercel-apscheduler/per-job-cache-records.feature.md b/changes/vercel-apscheduler/per-job-cache-records.feature.md new file mode 100644 index 00000000..c9cea15c --- /dev/null +++ b/changes/vercel-apscheduler/per-job-cache-records.feature.md @@ -0,0 +1,9 @@ +The managed job store now keeps one Runtime Cache record per declared job +instead of a single document holding every job. The declarations are the +index: reads enumerate the code-declared ids, and a record that is missing, +unreadable, or written for a different declared schedule is rebuilt from its +declaration at the point of use, with a log line making cache eviction +observable. Eviction and write races now cost at most one job's progress +instead of the whole population's, the per-item size limit no longer bounds +the job count, and takeover syncs to the new code's declarations lazily with +no reconciliation sweep or marker. diff --git a/integrations/vercel-apscheduler/README.md b/integrations/vercel-apscheduler/README.md index 588e440c..4b089b7d 100644 --- a/integrations/vercel-apscheduler/README.md +++ b/integrations/vercel-apscheduler/README.md @@ -150,10 +150,11 @@ can actually carry it: themselves: racing finishers compute the same canonical successor under the same idempotency key, and the queue accepts it once. An evicted document never strands the chain. -- **Code-declared jobs are durable because code is the backup.** Whenever the - store's documents are missing, reconciliation rewrites declared jobs from - the declarations. The store holds nothing code cannot restate: jobs are - immutable at runtime. +- **Code-declared jobs are durable because code is the backup.** The store + keeps one record per declared job, the declarations are the index, and a + record that is missing, unreadable, or written for a different declared + schedule is rebuilt from its declaration at the point of use. The store + holds nothing code cannot restate: jobs are immutable at runtime. - **Scheduler lifecycle flags are best-effort.** A `pause()` can be lost to cache eviction, after which traffic reactivates the scheduler. `pause()` publishes a queue-borne control message so the flag reaches the process diff --git a/integrations/vercel-apscheduler/SCHEDULER.md b/integrations/vercel-apscheduler/SCHEDULER.md index 22a90e61..e5e793de 100644 --- a/integrations/vercel-apscheduler/SCHEDULER.md +++ b/integrations/vercel-apscheduler/SCHEDULER.md @@ -99,13 +99,15 @@ evicted. Each guarantee therefore lives on something that can carry it: against the driver document rather than atomically with the write, so a demoted deployment's stale pass aborts, but a narrow read-write race remains within the documented best-effort envelope. -- **Code-declared jobs are durable because code is the backup.** - Reconciliation rewrites them from the declarations whenever the documents - are missing, and the store holds nothing code cannot restate: runtime - creation of new jobs is rejected. Runtime changes to declared jobs and - lifecycle flags are best-effort by declared policy; `pause()` additionally - publishes a queue-borne control message so the flag reaches the process - serving the chain even where cache state does not. +- **Code-declared jobs are durable because code is the backup.** The + declarations are the index: reads enumerate the declared job ids, and a + record that is missing, unreadable, or written for a different declared + schedule is rebuilt from its declaration at the point of use + (read-repair). The store holds nothing code cannot restate: jobs are + immutable at runtime. Lifecycle flags remain best-effort by declared + policy; `pause()` additionally publishes a queue-borne control message so + the flag reaches the process serving the chain even where cache state + does not. Under `vercel dev` the cache client falls back to per-process memory, so the integration becomes a zero-infrastructure development mode with the @@ -131,17 +133,19 @@ dirty_logical_time earliest candidate parked by a concurrent store write idle_expires_at preview idle deadline, when enabled ``` -Jobs live in a second document beside it, one record per job with a revision -counter; the takeover reconciliation marker shares the jobs document so -eviction clears them together. Documents are rewritten on every touch and -carry a long TTL, so only an abandoned namespace is reaped; LRU eviction is -survivable by design (see above). +Each declared job has one record of its own beside it, keyed by the job id, +holding the serialized job, its execution progress, a per-record revision, +and a fingerprint of the declared trigger. Records are rewritten on every +touch and carry a long TTL, so only an abandoned namespace is reaped; LRU +eviction is survivable by design (see above), and it now costs one job's +progress, never the population's. -Every record is a code declaration plus execution progress; the store holds -nothing else. A record the reconciling code cannot load is rewritten from -its declaration when the code still declares it and removed when it does -not; a record found unreadable while planning due jobs is sidelined until -the next sync repairs it. +The record's fingerprint ties it to the declaration that wrote it. A read +that finds the fingerprint stale — the code's declared schedule changed — +rebuilds the record and restarts its schedule; a matching fingerprint keeps +the record's progress. A record whose declared id no longer exists in code +is simply unreachable, because enumeration comes from the declarations, and +it ages out by TTL. ## Starting @@ -285,7 +289,7 @@ keys are unsupported because they bypass wake rearming and revision checks. Each persisted job has a monotonic revision. After executing a job, the wake updates or removes it only if the revision it read is still current, so a -concurrent reconciliation write wins instead of being overwritten by a late +concurrent repair write wins instead of being overwritten by a late handler. A stale wake for a schedule the declarations no longer produce may already @@ -445,21 +449,20 @@ hands over promptly. Alias routing is judged by the request host, so do not point a manually created alias at an old deployment of a scheduler project: requests through that alias would let the old deployment take the chain. -On takeover the new owner reconciles the -store against its own declarations, before planning any due jobs: a job the -code no longer declares is deleted and never runs, a changed trigger restarts -its schedule, and an unchanged job keeps its progress. A job whose persisted -record no longer loads under the new code (typically because its function -moved) is rewritten from the declaration and restarts its schedule. - -Reconciliation completes only once it converges. A revision race with a -concurrent owner write reruns the pass against fresh state, and only the -owner marks the sync as done, after a clean pass; a reconciliation that -cannot converge stays unmarked and retries on the next activation. In-flight -work is never interrupted: the demoted deployment's running job finishes or -dies with its instance and its late writes are fenced best-effort. -Jobs that run long should enqueue their work to another queue and return, so -a promote is never delayed behind them. +On takeover the store syncs to the new code's declarations through +read-repair, before any wake plans due jobs: a job the code no longer +declares is unreachable and never runs, because every read enumerates the +reading deployment's own declarations; a changed trigger is detected by its +record's fingerprint and restarts its schedule on first read; an unchanged +job keeps its progress; and a record that no longer loads under the new +code (typically because its function moved) is rewritten from the +declaration. There is no sweep to converge and no marker to stamp — every +read validates against code. + +In-flight work is never interrupted: the demoted deployment's running job +finishes or dies with its instance and its late writes are fenced +best-effort. Jobs that run long should enqueue their work to another queue +and return, so a promote is never delayed behind them. A takeover strands the previous owner's in-flight wake: it is consumed by the demoted deployment and acked as stale, and the new owner's chain starts @@ -487,12 +490,10 @@ Deleting a deployment prevents its Functions from receiving further work. | resume while an old wake runs | the old generation cannot reserve a successor | | crash before Queue send | pending token is republished | | old message after resume | generation check makes it stale | -| handler finishes after a concurrent reconcile write | revision check preserves the newer record | +| handler finishes after a concurrent repair write | revision check preserves the newer record | | takeover while a wake is in flight | the demoted deployment consumes it and acks it as stale | | the owner's wake message dies | the overdue wake is presumed lost and republished by the owner | -| takeover reconciliation races a demoted deployment's handler | the demoted write aborts on the ownership fence | -| reconciliation loses a revision race to a concurrent owner write | the pass reruns with fresh state; completion is marked only once converged | -| a deployment loses the namespace mid-reconciliation | it cannot stamp the marker, and the owner reconciles | +| a demoted deployment's read wants to repair a record | the owner fence skips the write; it serves a declaration-derived view | | concurrent first requests | one automatic generation and one start identity | | request arrives after explicit pause | idle deadline renews but state remains paused | | preview idle deadline expires before claim | message is stale and no job runs | diff --git a/integrations/vercel-apscheduler/tests/unit/test_apscheduler_cache.py b/integrations/vercel-apscheduler/tests/unit/test_apscheduler_cache.py index 3770fdbb..59bfbf32 100644 --- a/integrations/vercel-apscheduler/tests/unit/test_apscheduler_cache.py +++ b/integrations/vercel-apscheduler/tests/unit/test_apscheduler_cache.py @@ -398,41 +398,6 @@ def test_cache_driver_foreign_owner_is_fenced_without_takeover() -> None: assert theirs.owner_deployment() == "dpl_b" -def test_cache_driver_mark_reconciled_is_owner_fenced() -> None: - ours = cache_driver("dpl_a") - theirs = cache_driver("dpl_b") - store = CacheJobStore() - store.bind_namespace(scope="prj_test:production", scheduler_id="conformance") - ours.attach_store(store) - theirs.attach_store(store) - now = datetime.now(UTC) - - ours.start(now) - assert not theirs.mark_reconciled("dpl_b", now) - assert ours.reconciled_deployment() is None - - assert ours.mark_reconciled("dpl_a", now) - assert ours.reconciled_deployment() == "dpl_a" - - -def test_cache_reconcile_marker_shares_the_jobs_document_fate() -> None: - driver = cache_driver("dpl_a") - store = CacheJobStore() - store.bind_namespace(scope="prj_test:production", scheduler_id="conformance") - driver.attach_store(store) - now = datetime.now(UTC) - - driver.start(now) - assert driver.mark_reconciled("dpl_a", now) - assert driver.reconciled_deployment() == "dpl_a" - - # Evicting the jobs document must clear the marker with it, so a driver - # document kept fresh by bridge hops cannot vouch for a reaped store. - assert store.doc_key is not None - get_cache().delete(store.doc_key) - assert driver.reconciled_deployment() is None - - def test_cache_paused_document_is_touched_by_the_activation_hook() -> None: driver = cache_driver() now = datetime.now(UTC) @@ -504,8 +469,7 @@ def test_cache_end_to_end_start_activates_and_reserves_first_wake() -> None: assert snapshot.state == "running" assert snapshot.start_status == "active" - jobs, undecodable = adapter.coordinator.get_all_jobs_with_revisions() - assert undecodable == [] + jobs = adapter.coordinator.get_all_jobs_with_revisions() assert [job.id for job, _revision in jobs] == ["tick"] @@ -643,9 +607,9 @@ def test_cache_eviction_self_heals_from_the_next_wake() -> None: ) first_wake = WakeupPayload.from_payload(send.call_args.args[1]) - # Total eviction: both documents disappear. + # Total eviction: the driver document and the job record disappear. get_cache().delete(adapter.driver.key) - get_cache().delete(adapter.coordinator.store.doc_key) + get_cache().delete(adapter.coordinator.store._record_key("tick")) with patch( "vercel.integrations.apscheduler._adapter.vqs_sync.send", @@ -667,11 +631,11 @@ def test_cache_eviction_self_heals_from_the_next_wake() -> None: successor = WakeupPayload.from_payload(send.call_args.args[1]) assert successor.sequence == first_wake.sequence + 1 - jobs, _ = adapter.coordinator.get_all_jobs_with_revisions() + jobs = adapter.coordinator.get_all_jobs_with_revisions() assert [job.id for job, _revision in jobs] == ["tick"] -def test_cache_coordinator_cas_and_quarantine() -> None: +def test_cache_coordinator_cas_and_read_repair() -> None: _scheduler, adapter, start_payload = started_cache_scheduler() start_subscription = get_subscriptions()[0] with patch( @@ -688,29 +652,28 @@ def test_cache_coordinator_cas_and_quarantine() -> None: ) coordinator = adapter.coordinator - jobs, _ = coordinator.get_all_jobs_with_revisions() + jobs = coordinator.get_all_jobs_with_revisions() (job, revision) = jobs[0] assert not coordinator.cas_update_job(job, revision + 41) assert coordinator.cas_update_job(job, revision) - # Corrupt the persisted record: it must be reported undecodable, and due - # planning must quarantine rather than crash the chain. + # Corrupt the persisted record: the next read rebuilds it from its + # declaration instead of crashing or sidelining the chain. store = coordinator.store - doc = store._load() - doc["jobs"]["tick"]["state"] = "bm90LXBpY2tsZQ==" # b"not-pickle" - store._store(doc) - - jobs, undecodable = coordinator.get_all_jobs_with_revisions() - assert jobs == [] - assert [record[0] for record in undecodable] == ["tick"] + record = store._load_record("tick") + assert record is not None + record["state"] = "bm90LXBpY2tsZQ==" # b"not-pickle" + store._store_record("tick", record) - due = coordinator.get_due_jobs_with_revisions(datetime.now(UTC) + timedelta(days=1)) - assert due == [] - assert store._load()["jobs"]["tick"]["quarantined"] is True + jobs = coordinator.get_all_jobs_with_revisions() + assert [job.id for job, _revision in jobs] == ["tick"] + repaired = store._load_record("tick") + assert repaired is not None + assert repaired["state"] != "bm90LXBpY2tsZQ==" -def test_cache_jobs_document_eviction_alone_triggers_reconcile() -> None: +def test_cache_job_record_eviction_alone_is_read_repaired() -> None: _scheduler, adapter, start_payload = started_cache_scheduler() start_subscription, wake_subscription = get_subscriptions()[:2] @@ -728,11 +691,10 @@ def test_cache_jobs_document_eviction_alone_triggers_reconcile() -> None: ) first_wake = WakeupPayload.from_payload(send.call_args.args[1]) - # Only the jobs document is reaped; the driver document stays fresh - # (e.g. kept alive by bridge hops on a sparse schedule). The marker - # lives in the jobs document, so reconciliation must re-run. - get_cache().delete(adapter.coordinator.store.doc_key) - assert adapter.driver.reconciled_deployment() is None + # Only the job record is reaped; the driver document stays fresh + # (e.g. kept alive by bridge hops on a sparse schedule). The next read + # must rebuild the record from its declaration. + get_cache().delete(adapter.coordinator.store._record_key("tick")) with patch( "vercel.integrations.apscheduler._adapter.vqs_sync.send", @@ -748,7 +710,7 @@ def test_cache_jobs_document_eviction_alone_triggers_reconcile() -> None: ) assert _EXECUTIONS == ["ran"] - jobs, _ = adapter.coordinator.get_all_jobs_with_revisions() + jobs = adapter.coordinator.get_all_jobs_with_revisions() assert [job.id for job, _revision in jobs] == ["tick"] @@ -796,14 +758,13 @@ def test_cache_declared_add_rearms_a_dormant_chain() -> None: ) # Force dormancy (active generation, consumed watermark, no token) and - # evict the jobs document — a declaration restored by reconciliation + # evict the job record — a declaration restored onto a dormant chain # must mint the wake nothing else will. doc = adapter.driver._read() doc["current"] = None doc["last_sequence"] = 4 adapter.driver._write(doc, datetime.now(UTC)) - assert adapter.coordinator.store.doc_key is not None - get_cache().delete(adapter.coordinator.store.doc_key) + get_cache().delete(adapter.coordinator.store._record_key("tick")) declared = adapter._declared_jobs["tick"] adapter.coordinator.add_job(declared) diff --git a/integrations/vercel-apscheduler/tests/unit/test_apscheduler_integration.py b/integrations/vercel-apscheduler/tests/unit/test_apscheduler_integration.py index 1c0e5e28..3df1ecae 100644 --- a/integrations/vercel-apscheduler/tests/unit/test_apscheduler_integration.py +++ b/integrations/vercel-apscheduler/tests/unit/test_apscheduler_integration.py @@ -139,7 +139,6 @@ def __init__(self, deployment: str = "dpl_test") -> None: self.last_sequence = 0 self.owner: str | None = None self.owner_deployment_value: str | None = None - self.reconciled: str | None = None def owner_deployment(self) -> str | None: return self.owner_deployment_value @@ -221,16 +220,6 @@ def pause(self, now: datetime) -> bool: self.state = "paused" return changed - def reconciled_deployment(self) -> str | None: - return self.reconciled - - def mark_reconciled(self, deployment: str, now: datetime) -> bool: - del now - if self.owner_deployment_value != deployment: - return False - self.reconciled = deployment - return True - def repair_overdue_wake(self, now: datetime, *, grace_seconds: int = 600) -> bool: with self.lock: if ( @@ -616,7 +605,7 @@ def test_production_state_is_environment_scoped( assert adapter.scope == "prj_123:production" store = adapter.scheduler._jobstores["default"] - assert store.doc_key == f"aps:prj_123:production:{TEST_SCHEDULER_ID}:jobs" + assert store.key_prefix == f"aps:prj_123:production:{TEST_SCHEDULER_ID}:job:" def test_custom_environment_state_is_environment_scoped( @@ -744,15 +733,16 @@ def test_preview_state_stays_deployment_scoped( assert adapter.scope == "dpl_test" -def test_takeover_reconciles_declared_jobs( +def test_takeover_read_repair_syncs_records_to_the_new_declarations( monkeypatch: pytest.MonkeyPatch, ) -> None: - """A promote syncs the store to the new code's declarations. + """A promote syncs the store to the new code's declarations lazily. - Environment-scoped production state outlives deployments, so the new - deployment's first touch must delete undeclared jobs before any wake - plans them, restart changed schedules, and preserve progress for - unchanged ones. + Environment-scoped production records outlive deployments. Enumeration + comes from the reading deployment's own declarations, so a job the new + code no longer declares is unreachable before any wake plans it; a + changed schedule is rebuilt from its declaration on first read; an + unchanged job keeps its progress. """ monkeypatch.setenv("VERCEL_ENV", "production") monkeypatch.setenv("VERCEL_PROJECT_ID", "prj_123") @@ -769,10 +759,10 @@ def test_takeover_reconciles_declared_jobs( store: Any = first._jobstores["default"] kept_run_time = store.lookup_job("keep").next_run_time changed_run_time = store.lookup_job("changed").next_run_time - assert driver.reconciled == "dpl_test" + assert store.lookup_job("gone") is not None # A new deployment takes over the shared namespace and driver. Its own - # injected store reads the same cache documents. + # injected store reads the same cache records. _clear_scheduler_registrations() monkeypatch.setenv("VERCEL_DEPLOYMENT_ID", "dpl_two") second = BlockingScheduler(timezone=UTC) @@ -797,7 +787,8 @@ def test_takeover_reconciles_declared_jobs( assert shared.lookup_job("keep").next_run_time == kept_run_time assert shared.lookup_job("changed").next_run_time != changed_run_time assert shared.lookup_job("changed").trigger.interval == timedelta(hours=2) - assert driver.reconciled == "dpl_two" + # The undeclared job is unreachable through the new code's enumeration. + assert shared.lookup_job("gone") is None def test_implicit_activation_respects_chain_ownership( @@ -826,7 +817,6 @@ def test_implicit_activation_respects_chain_ownership( send.assert_called_once() assert driver.owner_deployment_value == "dpl_test" - assert driver.reconciled == "dpl_test" def test_stale_deployment_touches_are_inert( @@ -875,8 +865,12 @@ def test_stale_deployment_touches_are_inert( stale_adapter.ensure_local_started() - # No resurrection of "legacy" in the shared namespace. + # No resurrection of "legacy" in the shared namespace: the demoted + # deployment's reads see a declaration-derived view but write no record. assert {job.id for job in store.get_all_jobs()} == {"kept"} + stale_store: Any = stale._jobstores["default"] + assert {job.id for job in stale_store.get_all_jobs()} == {"legacy"} + assert stale_store._load_record("legacy") is None assert stale_adapter.publish_pending_wakeup() is None with pytest.raises(APSchedulerConfigurationError, match="no longer drives"): stale_adapter.prepare_runtime_mutation() @@ -1873,7 +1867,6 @@ def test_durable_job_requires_explicit_id() -> None: def test_cold_start_preserves_persisted_next_run_time() -> None: scheduler, adapter, driver = scheduler_with_driver() driver.owner_deployment_value = "dpl_test" - driver.reconciled = "dpl_test" persisted_time = datetime.now(UTC) + timedelta(hours=2) scheduler.add_job( diff --git a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_adapter.py b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_adapter.py index a12abb91..a966ffcf 100644 --- a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_adapter.py +++ b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_adapter.py @@ -27,7 +27,6 @@ STATE_RUNNING, STATE_STOPPED, BaseScheduler, - IntervalTrigger, JobEvent, JobSubmissionEvent, MaxInstancesReachedError, @@ -59,11 +58,6 @@ # Queue delivery rounds wake delays up to whole seconds and adds dispatch # latency, so a grace below this cannot be met and skips occurrences. MIN_QUEUE_MISFIRE_GRACE_SECONDS = 5 -# A reconciliation pass can lose a revision race to a concurrent owner write; -# rerunning with fresh state converges. Bounded so a mutation storm defers to -# the next activation instead of spinning. -RECONCILE_PASS_LIMIT = 3 - # One live adapter per durable identity. Two schedulers whose stores collapse # to the same identity would interleave one namespace, so the second claim # fails loudly instead. @@ -151,25 +145,6 @@ def get_adapter(scheduler: Any) -> SchedulerAdapter | None: return cast("SchedulerAdapter | None", getattr(scheduler, ADAPTER_ATTR, None)) -def _trigger_fingerprint(trigger: Any) -> tuple[str, str, str]: - """Digest a trigger into its user-declared, comparable schedule. - - ``IntervalTrigger`` without an explicit ``start_date`` auto-anchors at - declaration time, so that field would look changed on every deployment - and re-anchor unchanged schedules; it is excluded from the digest. - """ - state: Any = trigger.__getstate__() - if isinstance(state, dict) and type(trigger) is IntervalTrigger: - state = {key: value for key, value in state.items() if key != "start_date"} - return ( - type(trigger).__module__, - type(trigger).__qualname__, - repr(sorted(state.items(), key=lambda item: str(item[0]))) - if isinstance(state, dict) - else repr(state), - ) - - class SchedulerAdapter: """Turns one durable APScheduler instance into one fenced Queue driver.""" @@ -185,7 +160,6 @@ def __init__( self._scope: str | None = None self._registration_deferred = False self._declared_jobs: dict[str, Any] = {} - self._reconciled = False self._driver: Driver | None = None self._coordinator: JobCoordinator | None = None self._backend: Backend | None = None @@ -271,8 +245,8 @@ def _owns_namespace(self) -> bool: def _scope_outlives_deployments(self) -> bool: """Whether the namespace can outlive this code's view of it. - Always true on the cache backend: its documents are evictable in - every scope, so reconciliation-from-code is the durability story + Always true on the cache backend: its records are evictable in + every scope, so read-repair-from-code is the durability story regardless of scoping. A future durable backend would return ``self._scope != self._deployment`` here. """ @@ -325,7 +299,6 @@ def start(self, *, paused: bool = False) -> None: now, idle_timeout_seconds=self._preview_idle_timeout_seconds(), ) - self._reconcile_takeover(now) self._publish_start_if_needed(decision, now=now) if not decision.changed and decision.start_status == "active": self._rearm_wake_from_stores(now) @@ -360,7 +333,6 @@ def auto_activate( ) if not decision.owned: return False - self._reconcile_takeover(now) if decision.state != "running": self._pause_local() return True @@ -590,7 +562,6 @@ def ensure_local_started(self) -> None: self._bind_runtime() self._validate_durable_configuration() if self.scheduler.state != STATE_STOPPED: - self._reconcile_takeover(datetime.now(UTC)) return self._inject_inline_executor() previous = self._suppress_wakeup @@ -604,7 +575,6 @@ def ensure_local_started(self) -> None: raise finally: self._suppress_wakeup = previous - self._reconcile_takeover(datetime.now(UTC)) def _pause_local(self) -> None: if self.scheduler.state != STATE_RUNNING: @@ -759,7 +729,10 @@ def materialize_pending_job( MIN_QUEUE_MISFIRE_GRACE_SECONDS, ) if not (self.is_runtime_mutation or self.is_wake_mutation): - # Cold-start declarations are the reconciliation input on takeover. + # Cold-start declarations are the store's read-repair index, and + # repair may serialize them at any read, so complete the defaults + # upstream _real_add_job would otherwise fill later. + self._fill_declaration_defaults(job, jobstore_alias) self._declared_jobs[str(job.id)] = job if job.executor != "default": raise APSchedulerConfigurationError( @@ -767,9 +740,8 @@ def materialize_pending_job( ) if not (self.is_runtime_mutation or self.is_wake_mutation) and not self._owns_namespace(): # A stale deployment's cold start must not write declarations into - # a namespace another deployment drives; taking ownership runs the - # reconciliation that writes them instead. - self._fill_declaration_defaults(job, jobstore_alias) + # a namespace another deployment drives; read-repair writes them + # once ownership arrives. return False existing = jobstore.lookup_job(job.id) if existing is None: @@ -779,16 +751,13 @@ def materialize_pending_job( f'job "{job.id}" already exists in the durable store; ' "declare it with replace_existing=True" ) - if self.is_runtime_mutation or self.is_wake_mutation: - return True - self._fill_declaration_defaults(job, jobstore_alias) - return False + return self.is_runtime_mutation or self.is_wake_mutation def _fill_declaration_defaults(self, job: Any, jobstore_alias: str) -> None: - """Complete a declared job whose store write was skipped. + """Complete a declared job the way upstream ``_real_add_job`` would. - The skipped write leaves the object without the defaults upstream - ``_real_add_job`` would fill; reconciliation may persist it later. + Declarations serve as the store's read-repair input, which can + serialize them before upstream fills their defaults. """ replacements: dict[str, Any] = { key: value @@ -893,129 +862,9 @@ def _validate_materialized_jobs(self) -> None: f'job "{job.id}" in "default" must use the default executor' ) - def _reconcile_takeover(self, now: datetime) -> None: - """Sync code-declared jobs into a namespace another deployment wrote. - - Environment-scoped stores outlive deployments, so on the first touch - by a new deployment the code's declarations win: removed declarations - are deleted before any due planning, changed triggers restart their - schedule, unchanged jobs keep their progress, and a record that no - longer loads is rewritten from its declaration when the code still - declares it and removed when it does not. - """ - if not self._scope_outlives_deployments: - return - if not self._owns_namespace(): - return - # The durable marker is the authority; the in-process flag is only a - # cache of it. Consulting the marker first means a rollback onto a - # warm instance re-reconciles, and an evicted cache document heals. - if self.driver.reconciled_deployment() == self.deployment: - self._reconciled = True - return - self._reconciled = False - try: - for _ in range(RECONCILE_PASS_LIMIT): - if not self._reconcile_pass(now): - continue - if self.driver.mark_reconciled(self.deployment, now): - self._reconciled = True - return - except NamespaceFencedError: - # Ownership moved mid-reconciliation; the marker stays unset and - # the current owner reconciles instead. - self._logger.info( - 'Deployment "%s" lost the scheduler while reconciling; ' - "leaving the sync to the current owner", - self.deployment, - ) - return - self._logger.warning( - "Reconciliation kept losing revision races to concurrent writes; " - "it stays unfinished and retries on the next activation" - ) - - def _reconcile_pass(self, now: datetime) -> bool: - """Run one full declaration sync; False when a revision race dirtied it. - - A lost compare-and-set means a concurrent owner write moved a job - after this pass read it; the caller reruns against fresh state so the - sync always reflects the store it actually saw. - """ - clean = True - missing = dict(self._declared_jobs) - with self.scheduler._jobstores_lock: - jobs, undecodable = self.coordinator.get_all_jobs_with_revisions() - clean &= self._reconcile_undecodable(undecodable, missing, now) - for job, revision in jobs: - missing.pop(str(job.id), None) - declared = self._declared_jobs.get(str(job.id)) - if declared is None: - if self.coordinator.cas_remove_job(job.id, revision): - self._logger.info( - 'Removed job "%s": this deployment no longer declares it', - job.id, - ) - else: - clean = False - continue - if _trigger_fingerprint(declared.trigger) == _trigger_fingerprint(job.trigger): - next_run_time = job.next_run_time - else: - next_run_time = declared.trigger.get_next_fire_time( - None, - now.astimezone(self.scheduler.timezone), - ) - declared._modify(next_run_time=next_run_time) - if not self.coordinator.cas_update_job(declared, revision): - clean = False - for declared in missing.values(): - # Declared before this deployment owned the namespace, so the - # cold-start materialization deliberately did not write it. - self.coordinator.add_job(declared) - return clean - - def _reconcile_undecodable( - self, - undecodable: list[tuple[str, int]], - missing: dict[str, Any], - now: datetime, - ) -> bool: - """Repair or remove records this deployment's code cannot load.""" - clean = True - for job_id, revision in undecodable: - declared = missing.pop(job_id, None) - if declared is None: - if self.coordinator.cas_remove_job(job_id, revision): - self._logger.info( - 'Removed job "%s": this deployment no longer declares it', - job_id, - ) - else: - clean = False - continue - # The code still declares this job but its persisted record no - # longer loads (typically its function moved). The declaration is - # authoritative; restart its schedule under the new code. - declared._modify( - next_run_time=declared.trigger.get_next_fire_time( - None, - now.astimezone(self.scheduler.timezone), - ) - ) - if self.coordinator.cas_update_job(declared, revision): - self._logger.info( - 'Rewrote job "%s" from its declaration: the persisted ' - "definition no longer loads under this deployment", - job_id, - ) - else: - clean = False - return clean - def _rebase_before(self, activation_time: datetime) -> None: with self.scheduler._jobstores_lock: - jobs, _undecodable = self.coordinator.get_all_jobs_with_revisions() + jobs = self.coordinator.get_all_jobs_with_revisions() for job, revision in jobs: next_run_time = job.next_run_time if next_run_time is None or next_run_time >= activation_time: diff --git a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/_protocols.py b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/_protocols.py index b42c5380..fa047ff4 100644 --- a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/_protocols.py +++ b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/_protocols.py @@ -3,8 +3,9 @@ A backend supplies two collaborators to the adapter: - a *driver*: the lifecycle state machine (start/pause/resume, wake tokens, - ownership, the reconciliation marker), and -- a *job coordinator*: atomic-as-possible job writes coupled to wake rearming. + ownership), and +- a *job coordinator*: atomic-as-possible job reads and writes coupled to + wake rearming. """ from __future__ import annotations @@ -137,10 +138,6 @@ def repair_overdue_wake(self, now: datetime) -> WakeToken | bool | None: def owner_deployment(self) -> str | None: ... - def reconciled_deployment(self) -> str | None: ... - - def mark_reconciled(self, deployment: str, now: datetime) -> bool: ... - def renew(self, owner: str, now: datetime) -> bool: ... def release(self, owner: str) -> None: ... @@ -169,16 +166,12 @@ def get_due_jobs_with_revisions( now: datetime, ) -> list[tuple[Any, int]]: ... - def get_all_jobs_with_revisions( - self, - ) -> tuple[list[tuple[Any, int]], list[tuple[str, int]]]: ... + def get_all_jobs_with_revisions(self) -> list[tuple[Any, int]]: ... def cas_update_job(self, job: Any, expected_revision: int) -> bool: ... def cas_remove_job(self, job_id: str, expected_revision: int) -> bool: ... - def quarantine_job(self, job_id: str) -> None: ... - class BoundRuntime(Protocol): """What a backend hands the adapter after binding to a scheduler.""" diff --git a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/__init__.py b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/__init__.py index 517bd673..6965b2b5 100644 --- a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/__init__.py +++ b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/__init__.py @@ -10,10 +10,11 @@ accepts the publication once. Claims are best-effort filters that shrink, but cannot eliminate, duplicate wake *executions*; the contract is at-least-once. -- **Declared jobs are reconstructable, not durable.** Code is the backup: a - missing driver or job document is rebuilt from declarations by the existing - reconcile/materialize machinery, so eviction can only cost state that code - cannot restate (execution progress, lifecycle flags). +- **Declared jobs are reconstructable, not durable.** Code is the index and + the backup: reads enumerate the declared job ids, and a missing, unreadable, + or schedule-changed record is rebuilt from its declaration at the point of + use (read-repair), so eviction can only cost state that code cannot restate + (execution progress, lifecycle flags). - **Lifecycle flags are best-effort** by declared policy: read-merge-write with bounded retries; last writer wins. ``pause()`` additionally rides the queue as a control message. @@ -116,7 +117,6 @@ def bind(self, adapter: Any, *, scope: str, deployment: str) -> _Bound: scheduler_id=adapter.identity.scheduler_id, deployment=deployment, ) - driver.attach_store(store) coordinator = CacheJobCoordinator(store, driver, adapter) coordinator.install() return _Bound(driver=driver, coordinator=coordinator) diff --git a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_doc.py b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_doc.py index 6fc7911d..754af269 100644 --- a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_doc.py +++ b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_doc.py @@ -1,4 +1,4 @@ -"""Shared Runtime Cache document plumbing for the cache backend.""" +"""Shared Runtime Cache entry plumbing for the cache backend.""" from __future__ import annotations @@ -8,13 +8,13 @@ from ..._time import as_utc -# 35 days. Docs are rewritten on every touch (wakes, activation-hook runs), -# so an active, paused, or dormant scheduler on a traffic-serving deployment -# never expires; the TTL only reaps abandoned namespaces (and, with them, -# lifecycle flags — declared jobs come back from code). +# 35 days. Entries are rewritten on every touch (wakes, activation-hook +# runs), so an active, paused, or dormant scheduler on a traffic-serving +# deployment never expires; the TTL only reaps abandoned namespaces (and, +# with them, lifecycle flags — declared jobs come back from code). # LRU eviction is the space-based reaper; this is the time-based one. DOC_TTL_SECONDS = 35 * 24 * 3600 -_INDEX_MERGE_ATTEMPTS = 4 +_WRITE_ATTEMPTS = 4 __all__ = ["DOC_TTL_SECONDS"] diff --git a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_driver.py b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_driver.py index 3e296f18..9dc2418d 100644 --- a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_driver.py +++ b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_driver.py @@ -60,7 +60,6 @@ def __init__(self, *, scope: str, scheduler_id: str, deployment: str) -> None: self.deployment = deployment self.key = f"aps:{scope}:{scheduler_id}:driver" self.tag = f"aps:{scope}:{scheduler_id}" - self._store: Any = None def _read(self) -> dict[str, Any]: doc = get_cache().get(self.key) @@ -576,28 +575,6 @@ def owner_deployment(self) -> str | None: value = self._read().get("owner_deployment") return value if isinstance(value, str) and value else None - def attach_store(self, store: Any) -> None: - self._store = store - - def reconciled_deployment(self) -> str | None: - if self._store is None: - return None - value = self._store._load().get("reconciled_deployment") - return value if isinstance(value, str) and value else None - - def mark_reconciled(self, deployment: str, now: datetime) -> bool: - del now - # Owner fence, best-effort: a demoted straggler must not stamp the - # marker. The marker is written into the jobs document so eviction - # clears them together and the next wake re-runs reconciliation - # instead of trusting a reaped store. - if self._store is None or self._read().get("owner_deployment") != deployment: - return False - doc = self._store._load() - doc["reconciled_deployment"] = deployment - self._store._store(doc) - return True - def renew(self, owner: str, now: datetime) -> bool: return True diff --git a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_jobstore.py b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_jobstore.py index 0a9094d4..5a953f00 100644 --- a/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_jobstore.py +++ b/integrations/vercel-apscheduler/vercel/integrations/apscheduler/_backends/cache/_jobstore.py @@ -1,4 +1,12 @@ -"""Cache job store and coordinator: declared-only, best-effort.""" +"""Cache job store and coordinator: one record per declared job. + +The declarations are the index. Every read enumerates the code-declared job +ids, so a record nothing declares is unreachable (it ages out by TTL), and a +record that is missing, unreadable, or whose declared schedule changed is +rebuilt from its declaration at the point of use (read-repair). Writes are +owner-fenced best-effort; a demoted deployment's reads degrade to a +declaration-derived view without writing. +""" from __future__ import annotations @@ -6,11 +14,11 @@ import base64 import logging -import operator import pickle import random import time from datetime import datetime, timezone +from itertools import starmap from apscheduler.job import Job # type: ignore[import-untyped] from apscheduler.jobstores.base import ( # type: ignore[import-untyped] @@ -20,39 +28,50 @@ ) from apscheduler.util import ( # type: ignore[import-untyped] datetime_to_utc_timestamp, - utc_timestamp_to_datetime, ) from vercel.cache import get_cache -from ..._types import ( - APSchedulerConfigurationError, - NamespaceFencedError, -) -from ._doc import _INDEX_MERGE_ATTEMPTS, DOC_TTL_SECONDS +from ..._imports import IntervalTrigger +from ..._types import APSchedulerConfigurationError, NamespaceFencedError +from ._doc import _WRITE_ATTEMPTS, DOC_TTL_SECONDS from ._driver import CacheDriver LOGGER = logging.getLogger("vercel.integrations.apscheduler") UTC = timezone.utc -_UNCHANGED = object() +__all__ = ["CacheJobCoordinator", "CacheJobStore", "trigger_fingerprint"] + -__all__ = ["CacheJobCoordinator", "CacheJobStore"] +def trigger_fingerprint(trigger: Any) -> str: + """Digest a trigger into its user-declared, comparable schedule. + + ``IntervalTrigger`` without an explicit ``start_date`` auto-anchors at + declaration time, so that field would look changed on every deployment + and re-anchor unchanged schedules; it is excluded from the digest. + """ + state: Any = trigger.__getstate__() + if isinstance(state, dict) and type(trigger) is IntervalTrigger: + state = {key: value for key, value in state.items() if key != "start_date"} + detail = ( + repr(sorted(state.items(), key=lambda item: str(item[0]))) + if isinstance(state, dict) + else repr(state) + ) + return f"{type(trigger).__module__}:{type(trigger).__qualname__}:{detail}" class CacheJobStore(BaseJobStore): # type: ignore[misc] - """APScheduler job store over one Runtime Cache document. + """APScheduler job store over one Runtime Cache record per declared job. - All jobs live in a single JSON document so every mutation is one - read-merge-write; the write methods are replaced by the coordinator at - bind time. Scaling note: this bounds the practical job count by the - cache's value-size limit; sharding is a follow-up if real projects hit - it. + Reads enumerate the declared ids through the coordinator, so eviction, + takeover, and schedule changes heal per job instead of per population; + the write methods are replaced by the coordinator at bind time. """ def __init__(self) -> None: super().__init__() self.pickle_protocol = pickle.HIGHEST_PROTOCOL - self.doc_key: str | None = None + self.key_prefix: str | None = None self.tag: str | None = None def bind_namespace(self, *, scope: str, scheduler_id: str) -> None: @@ -60,29 +79,35 @@ def bind_namespace(self, *, scope: str, scheduler_id: str) -> None: namespace = getattr(self, "_vercel_apscheduler_namespace", None) if namespace is not None and namespace != expected: raise APSchedulerConfigurationError("job store is already bound to another scheduler") - self.doc_key = f"aps:{scope}:{scheduler_id}:jobs" + self.key_prefix = f"aps:{scope}:{scheduler_id}:job:" self.tag = f"aps:{scope}:{scheduler_id}" self._vercel_apscheduler_namespace = expected - def _load(self) -> dict[str, Any]: - if self.doc_key is None: + @property + def _coordinator(self) -> CacheJobCoordinator: + coordinator = self.__dict__.get("_vercel_apscheduler_coordinator") + if coordinator is None: raise APSchedulerConfigurationError("cache job store is not bound yet") - doc = get_cache().get(self.doc_key) - if not isinstance(doc, dict) or not isinstance(doc.get("jobs"), dict): - return {"revision_counter": 0, "jobs": {}} - normalized = { - "revision_counter": int(doc.get("revision_counter") or 0), - "jobs": dict(doc["jobs"]), - } - # The reconcile marker shares this document (and its eviction fate). - if isinstance(doc.get("reconciled_deployment"), str): - normalized["reconciled_deployment"] = doc["reconciled_deployment"] - return normalized + return coordinator # type: ignore[no-any-return] - def _store(self, doc: dict[str, Any]) -> None: - if self.doc_key is None: + def _record_key(self, job_id: str) -> str: + if self.key_prefix is None: raise APSchedulerConfigurationError("cache job store is not bound yet") - get_cache().set(self.doc_key, doc, {"ttl": DOC_TTL_SECONDS, "tags": [self.tag]}) + return f"{self.key_prefix}{job_id}" + + def _load_record(self, job_id: str) -> dict[str, Any] | None: + record = get_cache().get(self._record_key(job_id)) + return record if isinstance(record, dict) else None + + def _store_record(self, job_id: str, record: dict[str, Any]) -> None: + get_cache().set( + self._record_key(job_id), + record, + {"ttl": DOC_TTL_SECONDS, "tags": [self.tag]}, + ) + + def _delete_record(self, job_id: str) -> None: + get_cache().delete(self._record_key(job_id)) def _reconstitute_job(self, job_state: dict[str, Any]) -> Any: """Rebuild a Job exactly as upstream APScheduler stores do.""" @@ -96,58 +121,26 @@ def _decode(self, record: dict[str, Any]) -> Any | None: try: state = pickle.loads(base64.b64decode(record["state"])) return self._reconstitute_job(state) - except Exception: # noqa: BLE001 - any unpickling failure quarantines + except Exception: # noqa: BLE001 - any unpickling failure repairs return None - @staticmethod - def _run_time(record: dict[str, Any]) -> float | None: - value = record.get("next_run_time_ts") - return float(value) if value is not None else None - def lookup_job(self, job_id: str) -> Any | None: - record = self._load()["jobs"].get(str(job_id)) - if record is None: - return None - job = self._decode(record) - if job is not None: - job.id = str(job_id) - return job + entry = self._coordinator.entry(str(job_id)) + return entry[0] if entry is not None else None def get_due_jobs(self, now: datetime) -> list[Any]: - timestamp = datetime_to_utc_timestamp(now) - jobs = [] - for record in self._load()["jobs"].values(): - run_time = self._run_time(record) - if record.get("quarantined") or run_time is None or run_time > timestamp: - continue - job = self._decode(record) - if job is not None: - jobs.append((run_time, job)) - return [job for _, job in sorted(jobs, key=operator.itemgetter(0))] + return [job for job, _revision in self._coordinator.get_due_jobs_with_revisions(now)] def get_next_run_time(self) -> datetime | None: run_times = [ - run_time - for record in self._load()["jobs"].values() - if not record.get("quarantined") and (run_time := self._run_time(record)) is not None + job.next_run_time + for job, _revision in self._coordinator.get_all_jobs_with_revisions() + if job.next_run_time is not None ] - if not run_times: - return None - return utc_timestamp_to_datetime(min(run_times)) + return min(run_times, default=None) def get_all_jobs(self) -> list[Any]: - jobs = [] - for record in self._load()["jobs"].values(): - job = self._decode(record) - if job is not None: - jobs.append(job) - jobs.sort( - key=lambda job: ( - job.next_run_time is None, - job.next_run_time or datetime.max.replace(tzinfo=UTC), - ) - ) - return jobs + return [job for job, _revision in self._coordinator.get_all_jobs_with_revisions()] def add_job(self, job: Any) -> None: # pragma: no cover - replaced by install() raise APSchedulerConfigurationError("cache job store used before binding") @@ -163,13 +156,14 @@ def remove_all_jobs(self) -> None: # pragma: no cover - replaced by install() class CacheJobCoordinator: - """Couples the cache job store to its driver, best-effort. + """Couples per-job records to the driver: declared-only, read-repair. The store is immutable at runtime: every record is a code declaration plus execution progress, so eviction can never lose state that code and - the in-flight messages cannot restate. The revision counter and CAS - checks are read-merge-write rather than atomic, which shrinks but cannot - eliminate lost updates under concurrency. + the in-flight messages cannot restate. Revision checks are + read-merge-write rather than atomic, which shrinks but cannot eliminate + lost updates under concurrency — and a race now only ever involves one + job's record, never its neighbors. """ def __init__(self, store: CacheJobStore, driver: CacheDriver, adapter: Any) -> None: @@ -184,6 +178,22 @@ def install(self) -> None: self.store.remove_all_jobs = self.remove_all_jobs # type: ignore[method-assign] # ty: ignore[invalid-assignment] self.store.__dict__["_vercel_apscheduler_coordinator"] = self + # --- record plumbing ------------------------------------------------- + + def _declared(self) -> dict[str, Any]: + return dict(self.adapter._declared_jobs) + + def _owner_allows_writes(self) -> bool: + owner = self.driver.owner_deployment() + return owner is None or owner == self.driver.deployment + + def _check_fence(self, subject: str) -> None: + if not self._owner_allows_writes(): + raise NamespaceFencedError( + f'deployment "{self.driver.deployment}" no longer drives this ' + f"scheduler; the write to {subject} was fenced" + ) + def _record(self, job: Any, revision: int) -> dict[str, Any]: state = pickle.dumps(job.__getstate__(), self.store.pickle_protocol) next_run_time = getattr(job, "next_run_time", None) @@ -193,40 +203,120 @@ def _record(self, job: Any, revision: int) -> dict[str, Any]: datetime_to_utc_timestamp(next_run_time) if next_run_time is not None else None ), "revision": revision, - "quarantined": False, + "fingerprint": trigger_fingerprint(job.trigger), } - def _mutate(self, apply: Any, *, fenced: bool = True) -> Any: - """Read-merge-write with bounded retries against transient failures. + def _persist(self, job_id: str, record: dict[str, Any]) -> None: + """Write one record with bounded retries against transient failures.""" + last_error: Exception | None = None + for attempt in range(_WRITE_ATTEMPTS): + try: + self.store._store_record(job_id, record) + except Exception as exc: # noqa: BLE001 - cache I/O is best-effort + last_error = exc + time.sleep(random.uniform(0.02, 0.1) * (attempt + 1)) + else: + return + raise RuntimeError("cache job store write failed") from last_error - The owner fence is best-effort (checked against the driver document, - not atomically with the write), but it keeps the adapter's - ``NamespaceFencedError`` paths live: a demoted deployment's stale - pass aborts instead of resurrecting old declarations. - """ + def _erase(self, job_id: str) -> None: last_error: Exception | None = None - for attempt in range(_INDEX_MERGE_ATTEMPTS): - if fenced: - owner = self.driver.owner_deployment() - if owner is not None and owner != self.driver.deployment: - raise NamespaceFencedError( - f'deployment "{self.driver.deployment}" no longer drives ' - "this scheduler; the job-store write was fenced" - ) + for attempt in range(_WRITE_ATTEMPTS): try: - doc = self.store._load() - result = apply(doc) - if result is not _UNCHANGED: - self.store._store(doc) - except APSchedulerConfigurationError: - raise + self.store._delete_record(job_id) except Exception as exc: # noqa: BLE001 - cache I/O is best-effort last_error = exc time.sleep(random.uniform(0.02, 0.1) * (attempt + 1)) else: - return result + return raise RuntimeError("cache job store write failed") from last_error + # --- enumeration with read-repair ------------------------------------ + + def entry(self, job_id: str) -> tuple[Any, int] | None: + """Return ``(job, revision)`` for one declared id, repairing as needed.""" + declared = self._declared().get(str(job_id)) + if declared is None: + return None + return self._entry_for(str(job_id), declared) + + def _entry_for(self, job_id: str, declared: Any) -> tuple[Any, int]: + record = self.store._load_record(job_id) + if record is not None: + job = self.store._decode(record) + if job is not None and record.get("fingerprint") == trigger_fingerprint( + declared.trigger + ): + job.id = job_id + return job, int(record.get("revision") or 0) + return self._repair(job_id, declared, record) + + def _repair( + self, + job_id: str, + declared: Any, + stale_record: dict[str, Any] | None, + ) -> tuple[Any, int]: + """Rebuild one record from its declaration at the point of use. + + Restarting the schedule from now is deliberate: the record's own + progress is gone or belongs to a different declared trigger, and + recomputing from now skips the unobserved interval instead of + replaying it (a past-due date declaration does not re-fire). + + The write is owner-fenced and best-effort: a demoted deployment + still gets a declaration-derived view for local reads, but writes + nothing into the namespace it no longer drives. + """ + scheduler = getattr(self.store, "_scheduler", None) + now = datetime.now(scheduler.timezone if scheduler is not None else UTC) + state = declared.__getstate__() + state["next_run_time"] = declared.trigger.get_next_fire_time(None, now) + job = self.store._reconstitute_job(state) + job.id = job_id + revision = int((stale_record or {}).get("revision") or 0) + 1 + if self._owner_allows_writes(): + try: + self._persist(job_id, self._record(job, revision)) + except RuntimeError: + LOGGER.exception( + 'Could not persist the rebuilt record for job "%s"; ' + "serving the declaration-derived view", + job_id, + ) + else: + LOGGER.warning( + 'Rebuilt job "%s" from its declaration: its record was %s', + job_id, + ( + "missing (possible cache eviction)" + if stale_record is None + else "written for a different declared schedule" + ), + ) + return job, revision + + def get_due_jobs_with_revisions(self, now: datetime) -> list[tuple[Any, int]]: + timestamp = datetime_to_utc_timestamp(now) + return [ + (job, revision) + for job, revision in self.get_all_jobs_with_revisions() + if job.next_run_time is not None + and datetime_to_utc_timestamp(job.next_run_time) <= timestamp + ] + + def get_all_jobs_with_revisions(self) -> list[tuple[Any, int]]: + entries = list(starmap(self._entry_for, self._declared().items())) + entries.sort( + key=lambda entry: ( + entry[0].next_run_time is None, + entry[0].next_run_time or datetime.max.replace(tzinfo=UTC), + ), + ) + return entries + + # --- writes ----------------------------------------------------------- + def _reject_runtime_mutation(self, subject: str) -> None: """Refuse a runtime write: durable inputs are code and time. @@ -245,9 +335,8 @@ def _reject_runtime_mutation(self, subject: str) -> None: def _rearm(self, job: Any, *, always: bool = False) -> None: # Adds rearm unconditionally: a declaration restored onto a dormant - # chain (reconcile after jobs-doc eviction) must mint the wake - # nothing else will. rearm_wake's own guards make cold-start adds a - # no-op. + # chain must mint the wake nothing else will. rearm_wake's own guards + # make cold-start adds a no-op. if not (always or self.adapter.is_runtime_mutation): return next_run_time = getattr(job, "next_run_time", None) @@ -258,18 +347,15 @@ def _rearm(self, job: Any, *, always: bool = False) -> None: def add_job(self, job: Any) -> None: runtime = self.adapter.is_runtime_mutation or self.adapter.is_wake_mutation - - def apply(doc: dict[str, Any]) -> Any: - if str(job.id) in doc["jobs"]: - return "conflict" + job_id = str(job.id) + if self.store._load_record(job_id) is not None: + # Declarations are insert-if-absent; a runtime add of an existing + # id surfaces the conflict for upstream's replace_existing path, + # whose update is then rejected as a runtime mutation. if runtime: - return "declared-only" - doc["revision_counter"] += 1 - doc["jobs"][str(job.id)] = self._record(job, doc["revision_counter"]) - return None - - result = self._mutate(apply) - if result == "declared-only": + raise ConflictingIdError(job.id) + return + if runtime: # The store's contents must be reconstructable from code: an # evictable, per-region cache cannot durably hold the only copy # of a job nothing declares. @@ -279,120 +365,46 @@ def apply(doc: dict[str, Any]) -> Any: "in your own database, or publish a delayed queue message " "for one-shot work" ) - if result == "conflict": - # Declarations are insert-if-absent; a runtime add of an existing - # id surfaces the conflict so replace_existing can route through - # update_job, which mutates the declared job in place. - if runtime: - raise ConflictingIdError(job.id) - return + self._check_fence(f'job "{job_id}"') + self._persist(job_id, self._record(job, 1)) self._rearm(job, always=True) def update_job(self, job: Any) -> None: self._reject_runtime_mutation(f'update job "{job.id}"') - - def apply(doc: dict[str, Any]) -> Any: - record = doc["jobs"].get(str(job.id)) - if record is None: - return "missing" - doc["revision_counter"] += 1 - doc["jobs"][str(job.id)] = self._record(job, doc["revision_counter"]) - return None - - if self._mutate(apply) == "missing": + job_id = str(job.id) + record = self.store._load_record(job_id) + if record is None: raise JobLookupError(job.id) + self._check_fence(f'job "{job_id}"') + self._persist(job_id, self._record(job, int(record.get("revision") or 0) + 1)) self._rearm(job) def remove_job(self, job_id: str) -> None: self._reject_runtime_mutation(f'remove job "{job_id}"') - - def apply(doc: dict[str, Any]) -> Any: - if doc["jobs"].pop(str(job_id), None) is None: - return "missing" - doc["revision_counter"] += 1 - return None - - if self._mutate(apply) == "missing": + if self.store._load_record(str(job_id)) is None: raise JobLookupError(job_id) + self._check_fence(f'job "{job_id}"') + self._erase(str(job_id)) def remove_all_jobs(self) -> None: self._reject_runtime_mutation("remove jobs") - - def apply(doc: dict[str, Any]) -> Any: - doc["jobs"].clear() - doc["revision_counter"] += 1 - return None - - self._mutate(apply) - - def get_due_jobs_with_revisions(self, now: datetime) -> list[tuple[Any, int]]: - timestamp = datetime_to_utc_timestamp(now) - due: list[tuple[float, Any, int]] = [] - for job_id, record in self.store._load()["jobs"].items(): - run_time = CacheJobStore._run_time(record) - if record.get("quarantined") or run_time is None or run_time > timestamp: - continue - job = self.store._decode(record) - if job is None: - self.quarantine_job(job_id) - continue - due.append((run_time, job, int(record.get("revision") or 0))) - due.sort(key=operator.itemgetter(0)) - return [(job, revision) for _, job, revision in due] - - def get_all_jobs_with_revisions( - self, - ) -> tuple[list[tuple[Any, int]], list[tuple[str, int]]]: - jobs: list[tuple[Any, int]] = [] - undecodable: list[tuple[str, int]] = [] - for job_id, record in self.store._load()["jobs"].items(): - revision = int(record.get("revision") or 0) - job = self.store._decode(record) - if job is None: - undecodable.append((job_id, revision)) - continue - jobs.append((job, revision)) - jobs.sort( - key=lambda item: ( - item[0].next_run_time is None, - item[0].next_run_time or datetime.max.replace(tzinfo=UTC), - ), - ) - return jobs, undecodable + self._check_fence("the job store") + for job_id in self._declared(): + self._erase(job_id) def cas_update_job(self, job: Any, expected_revision: int) -> bool: - def apply(doc: dict[str, Any]) -> Any: - record = doc["jobs"].get(str(job.id)) - if record is None or int(record.get("revision") or 0) != expected_revision: - return False - doc["revision_counter"] += 1 - doc["jobs"][str(job.id)] = self._record(job, doc["revision_counter"]) - return True - - return bool(self._mutate(apply) is True) + self._check_fence(f'job "{job.id}"') + job_id = str(job.id) + record = self.store._load_record(job_id) + if record is None or int(record.get("revision") or 0) != expected_revision: + return False + self._persist(job_id, self._record(job, expected_revision + 1)) + return True def cas_remove_job(self, job_id: str, expected_revision: int) -> bool: - def apply(doc: dict[str, Any]) -> Any: - record = doc["jobs"].get(str(job_id)) - if record is None or int(record.get("revision") or 0) != expected_revision: - return False - del doc["jobs"][str(job_id)] - doc["revision_counter"] += 1 - return True - - return bool(self._mutate(apply) is True) - - def quarantine_job(self, job_id: str) -> None: - def apply(doc: dict[str, Any]) -> Any: - record = doc["jobs"].get(str(job_id)) - if record is None or record.get("quarantined"): - return _UNCHANGED - record["quarantined"] = True - return None - - self._mutate(apply, fenced=False) - LOGGER.error( - 'Quarantined APScheduler job "%s": its persisted definition can ' - "no longer be loaded by this deployment's code", - job_id, - ) + self._check_fence(f'job "{job_id}"') + record = self.store._load_record(str(job_id)) + if record is None or int(record.get("revision") or 0) != expected_revision: + return False + self._erase(str(job_id)) + return True