diff --git a/CHANGELOG.md b/CHANGELOG.md index 02114d9..3880a05 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,18 @@ test's rows and failed depending on order. Reported as reminders leaking, which is where it surfaces first because reminders outlive the message that created them. +- Document that a reminder is one named alarm per actor. `schedule` is keyed by + actor and reminder name, so scheduling a name that is already armed moves + that alarm rather than adding a second. The behaviour is deliberate and + matches Orleans and Durable Objects, but it was stated nowhere: an actor that + armed one reminder per queued item silently kept only the last, and the + earlier wake-ups never happened. The reminders guide now states the + uniqueness key and shows the one-alarm-many-items pattern to use instead. +- Add `solid_objects.reminder.replaced`, reported when a `schedule` call moves + an alarm already armed under the same name to a different time. It carries + the actor identity, reminder name, previous run time, and next run time, and + no arguments. Rescheduling to the same time reports nothing. The replacement + was previously indistinguishable from a first schedule. ## 0.10.2 - 2026-08-10 diff --git a/README.md b/README.md index 4dbfc04..ceba772 100644 --- a/README.md +++ b/README.md @@ -805,6 +805,62 @@ end Use `missed: :latest` to coalesce missed occurrences or `missed: :all` to enqueue each one. +### A reminder is one named alarm per actor + +The uniqueness key is `(actor, reminder name)`. Scheduling a name that is +already armed **moves the existing alarm** rather than adding a second one. The +database enforces this with a unique index on `(instance_id, name)`. + +This is the same model as Orleans reminders and Durable Objects alarms, and it +is what makes a reminder safe to re-arm from a handler that may run more than +once. It also means this is a data-loss bug: + +```ruby +# Wrong. Every entry overwrites the previous entry's alarm. +def add(entry:) + self.entries = entries + [ entry ] + schedule :deliver, at: entry.fetch("wait_until"), arguments: {} +end +``` + +Two entries leave one reminder. The earlier wake-up never happens, nothing +raises, and nothing is logged except a `solid_objects.reminder.replaced` event. + +Arm one alarm for the earliest item instead, and let the handler drain +everything now due before arming the next: + +```ruby +def add(entry:) + self.entries = (entries + [ entry ]).sort_by { |item| item.fetch("wait_until") } + arm_next +end + +def deliver + now = Time.current.to_i + due, pending = entries.partition { |item| item.fetch("wait_until") <= now } + due.each { |item| emit :send_push, **item.symbolize_keys } + self.entries = pending + arm_next +end + +private + +def arm_next + earliest = entries.first + return unless earliest + + schedule :deliver, at: Time.at(earliest.fetch("wait_until")), arguments: {} +end +``` + +`deliver` drains every due item rather than one, so a single alarm serves a +whole queue and a missed or coalesced occurrence cannot strand an entry. Use a +distinct reminder name only when you genuinely need two independent alarms on +one actor, such as `:deliver` and `:sweep`. + +Solid Objects has no `unschedule`. A reminder stops when its handler does not +re-arm it, and destroying an actor removes its reminders. + Self-scheduling actors should also have a low-frequency application reconciler. It may read `SolidObjects::Instance.states_for`, `.without_pending_work`, and `.orphaned`, but every repair must go through `async`. Never bulk-update actor diff --git a/docs/architecture.md b/docs/architecture.md index ebd379d..af19cc4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -439,6 +439,8 @@ A reminder record contains actor identity, a reminder name, target message, JSON schedule :expire, at: 30.minutes.from_now, arguments: {} ``` +Reminders are keyed by `(actor, reminder name)`, enforced by a unique index on `(instance_id, name)`. `schedule` is therefore an upsert: scheduling a name that is already armed moves that alarm instead of adding another, which is what makes re-arming safe from a handler that may run more than once. An actor needing several pending items should arm one alarm for the earliest and drain everything due when it fires, rather than one alarm per item; the [reminders guide](../README.md#a-reminder-is-one-named-alarm-per-actor) shows that pattern. A move that changes `next_run_at` emits `solid_objects.reminder.replaced`, because the replacement is otherwise indistinguishable from a first schedule. + When due, the scheduler locks the source instance and creates a normal mailbox row with an idempotency key derived from reminder ID and occurrence. The mailbox insert and reminder advancement commit atomically. The mailbox provides diff --git a/docs/database-schema.md b/docs/database-schema.md index 960e89a..40d25bd 100644 --- a/docs/database-schema.md +++ b/docs/database-schema.md @@ -77,7 +77,8 @@ Indexes: ### `reminders` Durable one-shot or recurring alarm definitions. Unique instance/name makes -rescheduling an actor-owned reminder deterministic. Status/next-run/ID drives +rescheduling an actor-owned reminder deterministic, which also means a second +`schedule` under the same name moves that alarm rather than adding one. Status/next-run/ID drives the due scan. Claim ownership references the process registry, and constraints limit recurrence intervals, missed-work policies, and statuses. diff --git a/docs/operations.md b/docs/operations.md index afd8931..e5f58e1 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -155,6 +155,15 @@ transaction rejection, commit-action start/completion/failure, effect and broadcast enqueue/completion, reminder enqueue, actor destruction/expiration, retention pruning, process cleanup, and supervisor lifecycle. +`solid_objects.reminder.replaced` reports a `schedule` call that moved an alarm +already armed under the same name on the same actor, carrying the actor +identity, reminder `name`, `previous_run_at`, and `next_run_at`. Reminders are +keyed by actor and name, so re-arming a name is an update rather than a second +alarm. That is deliberate, and it is silent: an actor that arms one reminder +per queued item keeps only the last, and the earlier wake-up never happens. +Watch this event if your actors schedule from a loop or from a handler that can +run more than once. Rescheduling to the same time reports nothing. + `solid_objects.component.refreshed` covers every authorized component refresh request. Its payload carries the actor identity, `component_name`, `component_key`, declared `dependencies`, `refresh_method`, the rendered diff --git a/lib/solid_objects/executor.rb b/lib/solid_objects/executor.rb index ba90e13..5b8dec1 100644 --- a/lib/solid_objects/executor.rb +++ b/lib/solid_objects/executor.rb @@ -90,8 +90,14 @@ def complete(result, observable_changes, state_changed:) reminder_intents = actor.drain_reminder_intents outbound_message_intents = actor.drain_outbound_message_intents enqueued_effects = [] + moved_reminders = [] activation.lease.fenced_transaction do |instance| + # A busy database makes the adapter retry this whole block, so an + # attempt that was rolled back must not leave its work in the lists the + # reporting below reads. Each attempt starts from empty. + enqueued_effects.clear + moved_reminders.clear claimed_message = matching_claim! locked_message = Message.lock.find(message.id) execute_commit_actions(commit_action_intents) @@ -107,7 +113,7 @@ def complete(result, observable_changes, state_changed:) completed_at: SolidObjects.database_adapter.database_now ) enqueued_effects.concat(enqueue_effects(locked_message, instance, effect_intents)) - schedule_reminders(instance, reminder_intents) + moved_reminders.concat(schedule_reminders(instance, reminder_intents)) enqueued_effects.concat( enqueue_actor_messages(locked_message, instance, outbound_message_intents) ) @@ -127,6 +133,9 @@ def complete(result, observable_changes, state_changed:) observable_name: ) end + moved_reminders.each do |moved| + SolidObjects.instrument(:"reminder.replaced", **moved) + end enqueued_effects.each do |effect| SolidObjects.instrument( :"effect.enqueued", @@ -211,10 +220,20 @@ def enqueue_effects(locked_message, instance, intents) end end - # @rbs (Instance, Array[Actor::ReminderIntent]) -> void + # A reminder is one named alarm per actor, so scheduling a name that is + # already armed moves it rather than adding a second. An actor that arms a + # reminder per queued item therefore keeps only the last, and nothing else + # about that is visible: the write succeeds and the earlier wake-up simply + # never happens. + # Moves are returned rather than reported here, so the report happens after + # the turn commits. A rolled back turn would otherwise announce an alarm + # that never moved, which is the opposite of the visibility this event + # exists to provide. + # @rbs (Instance, Array[Actor::ReminderIntent]) -> Array[Hash[Symbol, untyped]] def schedule_reminders(instance, intents) - intents.each do |intent| + intents.filter_map do |intent| reminder = Reminder.find_or_initialize_by(instance:, name: intent.name) + previous_run_at = reminder.next_run_at reminder.assign_attributes( actor_type: instance.actor_type, actor_id: instance.actor_id, @@ -227,10 +246,28 @@ def schedule_reminders(instance, intents) claimed_by: nil, claimed_at: nil ) + moved = moved_reminder_payload(reminder, previous_run_at) reminder.save! + moved end end + # Arguments are omitted deliberately: a reminder carries application data + # and this event exists to be logged. + # @rbs (Reminder, Time?) -> Hash[Symbol, untyped]? + def moved_reminder_payload(reminder, previous_run_at) + return nil unless previous_run_at + return nil if previous_run_at == reminder.next_run_at + + { + actor_type: reminder.actor_type, + actor_id: reminder.actor_id, + name: reminder.name, + previous_run_at:, + next_run_at: reminder.next_run_at + } + end + # @rbs (Message, Instance, Array[Actor::OutboundMessageIntent]) -> Array[Effect] def enqueue_actor_messages(locked_message, instance, intents) intents.map do |intent| diff --git a/sig/generated/lib/solid_objects/executor.rbs b/sig/generated/lib/solid_objects/executor.rbs index cbc6825..a1c27a5 100644 --- a/sig/generated/lib/solid_objects/executor.rbs +++ b/sig/generated/lib/solid_objects/executor.rbs @@ -45,8 +45,22 @@ module SolidObjects # @rbs (Message, Instance, Array[Actor::EffectIntent]) -> Array[Effect] def enqueue_effects: (Message, Instance, Array[Actor::EffectIntent]) -> Array[Effect] - # @rbs (Instance, Array[Actor::ReminderIntent]) -> void - def schedule_reminders: (Instance, Array[Actor::ReminderIntent]) -> void + # A reminder is one named alarm per actor, so scheduling a name that is + # already armed moves it rather than adding a second. An actor that arms a + # reminder per queued item therefore keeps only the last, and nothing else + # about that is visible: the write succeeds and the earlier wake-up simply + # never happens. + # Moves are returned rather than reported here, so the report happens after + # the turn commits. A rolled back turn would otherwise announce an alarm + # that never moved, which is the opposite of the visibility this event + # exists to provide. + # @rbs (Instance, Array[Actor::ReminderIntent]) -> Array[Hash[Symbol, untyped]] + def schedule_reminders: (Instance, Array[Actor::ReminderIntent]) -> Array[Hash[Symbol, untyped]] + + # Arguments are omitted deliberately: a reminder carries application data + # and this event exists to be logged. + # @rbs (Reminder, Time?) -> Hash[Symbol, untyped]? + def moved_reminder_payload: (Reminder, Time?) -> Hash[Symbol, untyped]? # @rbs (Message, Instance, Array[Actor::OutboundMessageIntent]) -> Array[Effect] def enqueue_actor_messages: (Message, Instance, Array[Actor::OutboundMessageIntent]) -> Array[Effect] diff --git a/test/integration/reminders_test.rb b/test/integration/reminders_test.rb index e75638a..0278da1 100644 --- a/test/integration/reminders_test.rb +++ b/test/integration/reminders_test.rb @@ -21,6 +21,80 @@ def expire end end + class QueueActor < SolidObjects::Actor + actor_type "reminder-queue" + + attribute :entries, default: -> { [] } + + def add(wait_until:) + self.entries = entries + [ wait_until ] + schedule :deliver, at: Time.at(wait_until).utc, arguments: {} + end + + def arm(name:, wait_until:) + schedule name.to_sym, at: Time.at(wait_until).utc, arguments: {} + end + + def deliver + self.entries = [] + end + end + + # The pattern the reminders guide recommends instead of one alarm per item. + # It is documented, so it is executed here rather than only read. + class QueueDrainActor < SolidObjects::Actor + actor_type "reminder-drain" + + attribute :entries, default: -> { [] } + attribute :delivered, default: -> { [] } + + def add(wait_until:) + self.entries = (entries + [ wait_until ]).sort + arm_next + end + + def deliver + now = Time.now.to_i + due, pending = entries.partition { |wait_until| wait_until <= now } + self.delivered = delivered + due + self.entries = pending + arm_next + end + + private + + def arm_next + earliest = entries.first + return unless earliest + + schedule :deliver, at: Time.at(earliest).utc, arguments: {} + end + end + + test "one alarm drains every due item and arms the next" do + reference = QueueDrainActor.ref("table") + due = 10.seconds.ago.to_i + later = 1.hour.from_now.to_i + reference.async(:add, wait_until: due) + reference.async(:add, wait_until: later) + worker = SolidObjects::Worker.new + worker.run_until_idle + scheduler = SolidObjects::ReminderScheduler.new + + assert scheduler.run_once, "the earliest item should have armed the alarm" + worker.run_until_idle + + state = SolidObjects::Instance.find_by!(actor_type: "reminder-drain").state + assert_equal [ due ], state.fetch("delivered") + assert_equal [ later ], state.fetch("entries") + reminder = SolidObjects::Reminder.find_by!(actor_type: "reminder-drain") + assert_equal later, reminder.next_run_at.to_i, + "the handler should have armed the next item before finishing" + ensure + scheduler&.stop + worker&.stop + end + test "persists a reminder in the actor commit" do message_reference = ExpiringActor.ref("one").async(:configure_expiration) worker = SolidObjects::Worker.new @@ -78,4 +152,124 @@ def expire assert_not actor.respond_to?(:remind, true) end + + # A reminder is one named alarm per actor, so a second schedule with the same + # name moves the existing alarm rather than adding one. An actor that arms a + # reminder per queued item therefore keeps only the last, which is silent + # data loss if the caller expected an alarm each. + test "a second schedule with the same name moves the existing reminder" do + reference = QueueActor.ref("table") + first = 1.hour.from_now.to_i + second = 2.hours.from_now.to_i + reference.async(:add, wait_until: first) + reference.async(:add, wait_until: second) + worker = SolidObjects::Worker.new + worker.run_until_idle + + reminders = SolidObjects::Reminder.where(actor_type: "reminder-queue") + assert_equal 1, reminders.count, + "two entries but one named reminder, which is the whole hazard" + assert_equal second, reminders.sole.next_run_at.to_i + end + + test "reminders with different names on one actor coexist" do + reference = QueueActor.ref("table") + reference.async(:arm, name: "deliver", wait_until: 1.hour.from_now.to_i) + reference.async(:arm, name: "sweep", wait_until: 2.hours.from_now.to_i) + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_equal %w[deliver sweep], + SolidObjects::Reminder.where(actor_type: "reminder-queue").order(:name).pluck(:name) + end + + test "the same reminder name on another actor is a separate reminder" do + QueueActor.ref("one").async(:add, wait_until: 1.hour.from_now.to_i) + QueueActor.ref("two").async(:add, wait_until: 2.hours.from_now.to_i) + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_equal 2, SolidObjects::Reminder.where(actor_type: "reminder-queue").count + end + + # The replacement is otherwise invisible, and finding it in production means + # finding it by noticing something never happened. + test "moving a reminder is instrumented" do + events = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.reminder.replaced") do |event| + events << event.payload + end + reference = QueueActor.ref("table") + reference.async(:add, wait_until: 1.hour.from_now.to_i) + reference.async(:add, wait_until: 2.hours.from_now.to_i) + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_equal 1, events.length, "only the replacement should report" + assert_equal "reminder-queue", events.sole.fetch(:actor_type) + assert_equal "table", events.sole.fetch(:actor_id) + assert_equal "deliver", events.sole.fetch(:name) + refute_equal events.sole.fetch(:previous_run_at), events.sole.fetch(:next_run_at) + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + worker&.stop + end + + test "scheduling reports moves to its caller rather than announcing them" do + executor = SolidObjects::Executor.allocate + instance = SolidObjects::Instance.create!( + actor_type: "reminder-queue", + actor_id: "unit", + state: {}, + state_version: 1 + ) + intent = SolidObjects::Actor::ReminderIntent.new( + name: "deliver", + at: 1.hour.from_now, + arguments: {}, + interval_seconds: nil, + missed_policy: "latest" + ) + later = SolidObjects::Actor::ReminderIntent.new( + name: "deliver", + at: 2.hours.from_now, + arguments: {}, + interval_seconds: nil, + missed_policy: "latest" + ) + events = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.reminder.replaced") do |event| + events << event.payload + end + + assert_empty executor.send(:schedule_reminders, instance, [ intent ]), + "a first schedule moves nothing" + moves = executor.send(:schedule_reminders, instance, [ later ]) + + assert_equal 1, moves.length, "the move should be returned to the caller" + assert_equal "deliver", moves.sole.fetch(:name) + assert_empty events, + "scheduling must not announce a move that the enclosing turn may roll back" + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + end + + test "rescheduling to the same time reports nothing" do + events = [] + subscription = ActiveSupport::Notifications.subscribe("solid_objects.reminder.replaced") do |event| + events << event.payload + end + wait_until = 1.hour.from_now.to_i + reference = QueueActor.ref("table") + reference.async(:add, wait_until:) + reference.async(:add, wait_until:) + worker = SolidObjects::Worker.new + worker.run_until_idle + + assert_empty events, + "a reschedule that moves nothing is not the hazard worth reporting" + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + worker&.stop + end end