[apscheduler] store one cache record per declared job

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.

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

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 |

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)