Conversation
Implements store-level primitives for parallel background processing of the CCC message log. Partitions are discovered via AND semantics (messages must have all partition attributes) and fetched via conditional AND (each message matches all partition attributes it has). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
claim_next now builds guard conditions from each handled_type × partition key_pair combination, enabling deciders to detect concurrent writes at append time via store.append(events, guard: result[:guard]). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Guard conditions are now derived from each message class's declared payload attributes via Message.to_conditions(**partition_attrs). This avoids nonsensical conditions (e.g. CourseCreated × user_id) while still covering all handled_types for conflict detection. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Documents conditional AND fetch, ConsistencyGuard from claim_next, Message.to_conditions, cached payload_attribute_names, and the SQLite DISTINCT requirement. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Enable causal chain tracing across CCC messages, matching the pattern from Sourced::Message. Both IDs default to the message's own id via Plumb's prepare_attributes hook. Store schema and serialization updated to persist and round-trip the new fields. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
highest_position on the consumer group tracks the furthest position ever successfully acked (advanced in ack, never decreased). claim_next returns replaying: true when all returned messages are at or below this watermark, meaning they have been processed before (e.g. after an offset reset). First-time processing is never replaying. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Replace the plain Hash with an immutable ClaimResult value object (Data.define) for type safety and a cleaner API. Defined in store.rb alongside the Store class. Tests updated to use method access. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Implement the Decide/Evolve/React pattern for CCC's stream-less model. Deciders request history via context_for() conditions, Projectors evolve from claimed messages directly. Router orchestrates claim→handle→execute→ack with transactional action execution, partial batch ACK, and error recovery. New modules: Actions (Append/Sync), Consumer, Evolve, React, Sync. Store#read now returns ReadResult data struct. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Decider now pre-correlates events with the command, then passes correlated events as source: to reaction Appends. This gives the correct causation chain: cmd → event → reaction message, with correlation_id tracing back to the command throughout. Append gains source: (override correlation source) and correlated: (skip re-correlation) options. Router integration tests verify exact causation_id and correlation_id at each link in the chain. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Add default empty handled_messages_for_evolve to CCC::Consumer so context_for works for reactors that just extend Consumer, define handled_messages, and implement handle_batch with manual action pairs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…partition reads Introduces Sourced::CCC.load(reactor_class, store, **partition_attrs) to load a reactor's evolved state from the store. Uses Store#read_partition for SQL-level AND filtering — a message is included only when every partition attribute it declares matches, avoiding loading irrelevant messages into memory. Guard's last_position covers the broader OR-context to prevent false concurrency conflicts. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…ier) Signal-driven dispatch for CCC reactors: Store notifies on append/resume, NotificationQueuer routes types to reactors via WorkQueue, Workers drain partitions in bounded loops. Reuses generic primitives (WorkQueue, CatchUpPoller, InlineNotifier) with CCC-specific Worker and Dispatcher. Also adds batch_size: to Store#claim_next and Router#handle_next_for. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Crashed workers leave partitions permanently claimed. Add a ccc_workers table with heartbeat upserts and a StaleClaimReaper that periodically releases claims from workers that stopped heartbeating. Wire the reaper into the Dispatcher alongside the existing notifier and catchup poller. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Mirrors Sourced::Supervisor but simpler: no separate HouseKeepers since StaleClaimReaper is already embedded in the CCC Dispatcher. Takes router + config kwargs, sets signal handlers, and spawns Dispatcher into an executor. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…ter, store, router, reset!) Wire CCC components (Supervisor, Dispatcher, Worker, Consumer, StaleClaimReaper) to pull defaults from CCC.config instead of Sourced.config, keeping executor on Sourced. Change CCC.load signature from positional store to keyword store: defaulting to global. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Uses Types::Interface to validate custom store objects implement the 12 required methods. Store instances and raw Sequel SQLite connections are still accepted directly. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Replace NullGroup with GroupUpdater that accumulates stop/retry mutations for atomic persistence. Add Store#updating_consumer_group to load, yield, and persist group state. Gate claim_next on retry_at so retries are honoured at the database level. Clear retry_at and error_context in start_consumer_group. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Encapsulates the validate → load history → decide → append → ACK flow into a single call. Returns a HandleResult supporting destructuring: cmd, reactor, events = CCC.handle!(cmd, MyDecider) Adds Store#advance_offset to move consumer group offsets without a prior claim, so background workers skip already-handled commands. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
stop_consumer_group, start_consumer_group, reset_consumer_group, and consumer_group_active? now accept either a String or any object responding to #group_id (e.g. a reactor class). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
So that framework integrations can register their own blocks that will be re-applied on fork
Monkey-patch with Sourced-specific methods for now. Rethink this
Also validate that a queue_mode reactor owns its message types
Enable non-Sourced reactors (e.g. Sidereal Commanders) to run on Sourced's
runtime, and make delete-on-ack an explicit, flag-driven concern.
Actions become inert signals:
- Reactors return plain `{type: :append|:schedule|:sync|:after_sync|:ack, ...}`
hashes (or Sourced::Actions value objects, which now `deconstruct_keys` to the
same shape). A new Sourced::ActionRunner is the sole code that touches the
store; Actions#execute is removed. Third parties emit signals without a Sourced
dependency.
- A `delete: true` flag on a signal is the ONLY thing that deletes a handled
message. Deletion is never implied by partitioning or exclusivity.
Duck-typed reactor protocol:
- Only `handled_messages` and `handle_claim` are required. ReactorDefaults.apply
defines any missing optional class-methods (`group_id` -> class name,
`partition_keys`, `exclusive?`, `on_exception`, `context_for`, lifecycle hooks)
directly on the reactor, only where absent, so the reactor stays a real class
(Injector signature reflection keeps working) and its own definitions win.
Queue partitioning via one engine:
- `exclusive` marks sole ownership (routing validation); it may omit
`partition_by` to get an id-partitioned queue (one partition per message),
reusing the existing offsets/key_pairs machinery. A reactor must otherwise
declare `partition_by` — forgetting it raises, never silently deletes.
- Store#append gains `index_by: :payload|:id`; the runner resolves it per target
message type. Reaper prunes drained queue offsets and orphaned id key_pairs.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Two gaps found in code review of the durable-messaging feature: 1. index_by was only applied by ActionRunner, so messages appended via Sourced.handle!, DurableWorkflow, or scheduled-message promotion (update_schedule!) were indexed by payload even when destined for an id-partitioned queue reactor — making them unclaimable (notably breaking future-scheduled queue commands). Fix: the Store now owns the per-type index basis, derived from each group's partitioning at register_consumer_group (handled_types:). append() resolves the basis per message, so every append path indexes consistently. index_by: stays as an optional override. This also removes Router#index_basis_for / @index_basis and ActionRunner's index_resolver + per-basis grouping. 2. Deletion is decoupled from exclusivity, but the offset reaper only reaped delivery_mode='queue' offsets, so a non-exclusive (log) reactor that deleted via a :delete action leaked drained offsets + orphan key_pairs. Fix: release_empty_queue_offsets -> release_drained_offsets reaps any unclaimed partition offset whose partition has no remaining messages, regardless of delivery mode. Safe because retain reactors never drain a partition (their messages persist). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The register guard only required `exclusive` when partition_by was omitted, so a reactor could explicitly declare `partition_by :id` while non-exclusive. That still id-indexes its message types (@type_index_basis[type] = :id) but skips exclusivity validation — if another reactor handled the same type, its messages became unclaimable (id-indexed but the co-consumer expects payload keys), silently breaking multi-consumer fan-out. Gate the guard on the effective partition keys (== [:id]) instead of "declared empty", so both the implicit (omitted) and explicit (`partition_by :id`) cases require exclusive. Restores the invariant: a type is only id-indexed when its lone exclusive owner is id-partitioned. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ck queues Document the durable-messaging additions: inert action signals + ActionRunner, the duck-typed reactor protocol via ReactorDefaults, exclusive/id-partitioned delete-on-ack queues, per-type index basis, multi-consumer fan-out, and the renamed/added store methods (append index_by, register_consumer_group handled_types/exclusive, release_drained_offsets, prune_orphan_key_pairs). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Partitioning by message id was keyed on the name "id", which shadows a legitimate user payload attribute named `id` — a user could write `partition_by :id` (or add an `id` payload field) expecting to partition by their own value, and silently get Message#id instead. Rename the reserved key to :__id / "__id" (unlikely to collide) across the router (effective_partition_keys default + exclusive guard) and store (effective_keys key name, id_partitioned?, register_consumer_group basis detection). :id is now a normal payload attribute name again. The internal index-basis symbol (:id vs :payload) is unchanged. Temporary rename; the stringly-typed magic key is worth replacing with a first-class flag later. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
delivery_mode became nearly vestigial after the review fixes: the offset reaper dropped its delivery_mode filter, leaving only reset_consumer_group's "no-op for queue groups" guard reading it. Exclusivity is already cached in-memory per group (@registered_groups), so derive the guard from that instead and drop the persisted column. Bonus: this also fixes existing databases created before delivery_mode was added. register_consumer_group no longer references the column, so a DB whose consumer_groups table lacks it (install! only create_table?s, never alters) no longer crashes on the INSERT — previously an opaque per-worker crash/respawn loop. Removes the column from the migration, from register_consumer_group's INSERT, and the three column-assertion specs (exclusivity is observable via the reset no-op behavior, which is still tested). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…e public Store#schedule_messages and :schedule action
…ueues README lagged behind the queue-mode work. Fix the now-broken "Delayed messages" example (schedule_messages is private — use append with a future-dated message) and add sections for the duck-typed reactor protocol (handled_messages/handle_claim + signals + ReactorDefaults) and delete-on-ack queues (exclusive + delete:true ack + id-partitioning). Correct the CLAUDE.md delete-on-ack example: a Decider's command DSL never emits delete:true, so show a Consumer-based queue overriding handle_claim. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Durable-messaging runtime: signal actions, duck-typed reactors, and delete-on-ack queues
For transactions that SELECT and then UPDATE, using just BEGIN starts a DEFERRED transaction (it takes a read snapshot of the last known WAL frame). Then the UPDATE or INSERT tries to promote the transaction to IMMEDIATE (get the global write lock), but if data has been inserted in the meantime (ie the WAL has new frames), the snapshot the process had is stale, so it raises BUSY_SNAPSHOT regardless of the BUSY_TIMEOUT config. The solution is to start these transaction in explicit IMMEDIATE mode. Also in this commit: make sure to add the relevant PRAGMAs to every connection checkout out from the pool. Some like foreign_keys = ON apparently need to be applied to every connection.
…lication. Sequel already applies per-connection
claim_next opened with a short-circuit that returns nil when types_max_pos <= last_nil_types_max_pos. That watermark is derived from handled_types, which for a Decider excludes its evolve types, and it is never invalidated by append or release. Under a concurrent append burst, a worker can poll nil (a sibling held the then-pending claims, or freshly-appended offsets weren't yet visible to its snapshot) and latch last_nil_types_max_pos at the max command position while eager offsets for those very commands are still pending. Because processing those commands only ever emits events — never a new command — types_max_pos never rises above the latch, so every subsequent poll short-circuits and the partitions are stranded forever with no error. For eager groups the offsets table is the authoritative record of claimable work, so scan it directly via find_and_claim_partition and skip the watermark entirely. The short-circuit stays on the legacy path, where offsets are created on demand and the messages table is the source of truth. types_max_pos is now computed only there, so idle eager groups also skip the MAX(position) query per poll. Adds a regression test that pins last_nil_types_max_pos at a pending eager offset's position and asserts claim_next still claims it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
It looped over every unclaimed offset in Ruby and ran a separate pending subquery per offset — N round-trips per poll. That was tolerable while the eager path short-circuited on last_nil_types_max_pos, but now that the eager path scans on every poll (see previous commit), an idle group with many partitions paid N queries each catch-up tick. Replace the loop with one query: a correlated EXISTS applies the AND-match (a message qualifies only when indexed under every one of the offset's key_pairs) and ORDER BY last_position LIMIT 1 picks the same offset the loop would have. EXPLAIN QUERY PLAN confirms it is fully index-backed (offsets cg/claimed index, message_type index, offset/message key_pair PKs). The guarded UPDATE ... WHERE claimed = 0 still handles the lost-race-to-a-sibling case, so semantics are unchanged — covered by the existing #claim_next specs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Removing the nil-claim short-circuit from the eager path (two commits ago) fixed the stranding bug but left every poll scanning all unclaimed offsets — O(partitions) per catch-up tick even when the group is fully idle. Bring the short-circuit back, correctly. For partitioned groups, last_nil_types_max_pos now holds a sign-encoded scan version: >= 0 means claimable work may have changed since the last empty scan, < 0 means an empty scan completed at version -value and nothing has changed since — so claim_next returns nil in O(1). Invalidation lives in the database as two triggers installed with the schema, so it holds for every writer — any process (including ones that never registered the group), the scheduled-message promoter, stale-claim reaping, or raw SQL: - AFTER INSERT ON messages: any new message may create claimable work, bump every partitioned group. - AFTER UPDATE OF claimed (1 -> 0) ON offsets: releasing a claim (ack, ack_and_delete, release, stale-claim reaping) can expose a batch remainder a sibling's nil scan skipped while the partition was claimed. claim_next latches clean via compare-and-set against the version it read before scanning; abs(value) + 1 keeps versions strictly monotonic, so a concurrent bump during the scan always misses the CAS (no ABA) and the group stays dirty. start_consumer_group resets the latch so resumed groups rescan. Legacy (NULL partition_by) groups are excluded from the triggers and keep the column's original watermark semantics. Benchmark, 5000 caught-up partitions: idle poll 15ms dirty scan once, then 0.07-0.10ms latched; one append re-opens the gate and is claimed. Verified end-to-end in the seats app: a full-speed 36-command burst (the original stranding scenario) drains completely with no errors and both groups settle latched. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…dically The claim scan's query plan is stats-dependent: without sqlite_stat1 the planner drives the pending-work EXISTS from the low-selectivity message_type index, degrading an idle scan to ~O(offsets x messages) — in practice a hang on large stores (a 30k-offset synthetic never completed). With stats it picks the key_pair-first plan, fully index-backed. Nothing ever ran ANALYZE. Add Store#optimize!: a bounded ANALYZE (analysis_limit = 400) with both statements pinned to one pooled connection, since analysis_limit is per-connection and would otherwise silently not apply. Run it from install! so every boot seeds stats, and from the StaleClaimReaper loop (default every 3600s) so stats track log growth. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Configurable codec, used by Store
…st Plumb 0.2.0.beta.2
Message codecs: native Ruby types in messages, JSON at the store boundary
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This branch replaces the original actor/stream-based architecture with the CCC (Command-Context Consistency) design, promoting it to the top-level
Sourcednamespace.Decider,Projector,DurableWorkflow, and plainConsumerreactors, all supporting#handle_batchSourced::Store(with Sequel migration support) replacingSequelBackend/PGBackend/SQLiteBackend/TestBackendDispatcher,Worker,StaleClaimReaper,ScheduledMessagePoller, consumer-group lifecycle hooksCommandContextwith per-message /anyhooks,Sourced.handle!for synchronous dispatchActor, old backends, pubsub modules, Rails generators,Unit, old handler/consumer testsexamples/app/demonstrating the new APIstore_spec.rbalone has 2400+ lines)