Merge pull request #41 from cardmagic/feat/keyed-reminders

Let a reminder be named for the item it waits on

Author: cardmagic

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## Unreleased
+
+- Accept a `key:` on `schedule`, naming a reminder for the item it is waiting
+  on rather than for its operation, so one actor can hold an alarm per queued
+  item. Scheduling the same key again moves that item's alarm and leaves the
+  others alone. Without a key the name is still the operation, so existing
+  reminders keep their names and their coalescing behaviour. A reminder
+  operation may no longer hold the colon that separates a key, which keeps
+  keyed and unkeyed names disjoint, and the length is checked on the composed
+  name rather than the key alone.
+
 ## 0.13.2 - 2026-08-16
 
 - Add an authorized `SolidObjects.administration.processes` query for

README.md

@@ -891,7 +891,7 @@ 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:
+once. Without a key the name is the operation, so this is a data-loss bug:
 
 ```ruby
 # Wrong. Every entry overwrites the previous entry's alarm.
@@ -904,8 +904,38 @@ 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:
+### An alarm per item, with `key:`
+
+Pass `key:` when an actor is waiting on several things at once. The key is your
+own identifier for the item, and it names that item's alarm, so each item gets
+one:
+
+```ruby
+def add(entry:)
+  self.entries = entries + [ entry ]
+  schedule(at: entry.fetch("wait_until"), key: entry.fetch("id")).deliver
+end
+```
+
+Two entries now leave two reminders. Scheduling the same key again moves that
+item's alarm and leaves the others alone, which is what makes a keyed reminder
+as safe to re-arm as an unkeyed one. The operation still decides which handler
+runs; the key only decides which alarm is which.
+
+A key must be non-empty, and the name it becomes must fit the 191-character
+column, which is checked on the composed name rather than the key alone so a
+long operation and a short key are caught too.
+
+The key is separated from the operation by a colon, so an operation may not hold
+one. Otherwise an unkeyed `deliver:item` and a `deliver` keyed `item` would be
+one name, and the second would silently take the first one's alarm. A key may
+hold colons of its own, because the operation before the first one cannot.
+
+### One alarm for a whole queue
+
+A key per item is not always what you want. An actor that only ever needs to
+know "what is next" can keep one alarm and let the handler drain everything now
+due before arming the next:
 
 ```ruby
 def add(entry:)
@@ -931,10 +961,10 @@ def arm_next
 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`.
+That costs one reminder row instead of one per item, and a coalesced occurrence
+cannot strand an entry because the handler drains by time rather than by alarm.
+Prefer it when the queue is large and the items are interchangeable; prefer
+`key:` when an item needs its own alarm that can be moved on its own.
 
 Solid Objects has no `unschedule`. A reminder stops when its handler does not
 re-arm it, and destroying an actor removes its reminders.

lib/solid_objects/actor.rb

@@ -4,7 +4,11 @@ module SolidObjects
   class Actor
     EffectIntent = Data.define(:name, :arguments, :success_operation, :failure_operation)
     CommitActionIntent = Data.define(:name, :arguments)
-    ReminderIntent = Data.define(:name, :at, :arguments, :interval_seconds, :missed_policy)
+    # The reminders table holds a name in 191 characters.
+    REMINDER_NAME_LIMIT = 191
+    REMINDER_KEY_SEPARATOR = ":"
+
+    ReminderIntent = Data.define(:name, :operation, :at, :arguments, :interval_seconds, :missed_policy)
     OutboundMessageIntent = Data.define(:actor_type, :actor_id, :operation, :arguments, :available_at, :idempotency_key)
 
     class << self
@@ -197,8 +201,13 @@ def commit_action(name, **arguments)
       nil
     end
 
-    # @rbs (at: Time, ?every: Numeric?, ?missed: Symbol | String) -> OperationDispatcher
-    def schedule(at:, every: nil, missed: :latest)
+    # A reminder is identified by its name, and without a key that name is the
+    # operation, so one actor holds one alarm per operation. A key gives an actor
+    # an alarm per item it is waiting on, which is what an actor holding a queue
+    # of scheduled work needs; the key is the caller's own identifier for the
+    # item, and scheduling the same key again moves that item's alarm.
+    # @rbs (at: Time, ?every: Numeric?, ?missed: Symbol | String, ?key: (String | Symbol | Integer)?) -> OperationDispatcher
+    def schedule(at:, every: nil, missed: :latest, key: nil)
       interval_seconds = every&.to_f
       if interval_seconds && !interval_seconds.positive?
         raise ArgumentError, "reminder interval must be positive"
@@ -207,13 +216,15 @@ def schedule(at:, every: nil, missed: :latest)
       unless %w[all latest].include?(missed_policy)
         raise ArgumentError, "missed reminder policy must be all or latest"
       end
+      reminder_key = validated_reminder_key(key)
 
       OperationDispatcher.new(
         actor_type: self.class.actor_type,
         handlers: self.class.definition.messages
       ) do |operation, arguments|
         ReminderIntent.new(
-          name: operation.to_s,
+          name: reminder_name(operation:, key: reminder_key),
+          operation: operation.to_s,
           at:,
           arguments: Serialization.dump(arguments),
           interval_seconds:,
@@ -225,6 +236,45 @@ def schedule(at:, every: nil, missed: :latest)
       end
     end
 
+    # @rbs ((String | Symbol | Integer)?) -> String?
+    def validated_reminder_key(key)
+      return nil if key.nil?
+
+      reminder_key = key.to_s
+      raise ArgumentError, "reminder key must not be empty" if reminder_key.empty?
+
+      reminder_key
+    end
+
+    # A keyed name is the operation, a colon, and the key, so an operation
+    # holding a colon of its own would make two different schedules produce one
+    # name: an unkeyed "deliver:item" and a "deliver" keyed "item" would share a
+    # row, and the second would silently take the first one's alarm. Refusing a
+    # colon in the operation keeps unkeyed names free of colons, which leaves
+    # the two kinds of name disjoint and lets a key hold colons of its own.
+    #
+    # The length is checked on the composed name rather than the key alone,
+    # because a long operation and a short key can exceed the column just as
+    # easily as the reverse. Both are refused here rather than at the insert,
+    # once the turn is already doing work.
+    # @rbs (operation: Symbol | String, key: String?) -> String
+    def reminder_name(operation:, key:)
+      operation_name = operation.to_s
+      if operation_name.include?(REMINDER_KEY_SEPARATOR)
+        raise ArgumentError,
+          "reminder operation #{operation_name.inspect} must not contain #{REMINDER_KEY_SEPARATOR.inspect}"
+      end
+      return operation_name if key.nil?
+
+      name = "#{operation_name}#{REMINDER_KEY_SEPARATOR}#{key}"
+      if name.length > REMINDER_NAME_LIMIT
+        raise ArgumentError,
+          "reminder name #{name.length} characters exceeds the #{REMINDER_NAME_LIMIT} the database holds"
+      end
+
+      name
+    end
+
     # @rbs (Reference, ?available_at: Time?, ?idempotency_key: String?) -> OperationDispatcher
     def send_to(reference, available_at: nil, idempotency_key: nil)
       actor_class = SolidObjects.registry.fetch(reference.actor_type)

lib/solid_objects/executor.rb

@@ -223,10 +223,10 @@ def enqueue_effects(message:, instance:, intents:)
     end
 
     # 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.
+    # already armed moves it rather than adding a second. Without a key that
+    # name is the operation, so an actor arming a reminder per queued item keeps
+    # only the last; passing schedule a key gives each item its own name and so
+    # its own alarm.
     # 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
@@ -239,7 +239,7 @@ def schedule_reminders(instance, intents)
         reminder.assign_attributes(
           actor_type: instance.actor_type,
           actor_id: instance.actor_id,
-          operation: intent.name,
+          operation: intent.operation,
           arguments: intent.arguments,
           next_run_at: intent.at,
           interval_seconds: intent.interval_seconds,

sig/generated/lib/solid_objects/actor.rbs

@@ -32,9 +32,16 @@ module SolidObjects
       def members: () -> [ :name, :arguments ]
     end
 
+    # The reminders table holds a name in 191 characters.
+    REMINDER_NAME_LIMIT: ::Integer
+
+    REMINDER_KEY_SEPARATOR: ::String
+
     class ReminderIntent < Data
       attr_reader name(): untyped
 
+      attr_reader operation(): untyped
+
       attr_reader at(): untyped
 
       attr_reader arguments(): untyped
@@ -43,12 +50,12 @@ module SolidObjects
 
       attr_reader missed_policy(): untyped
 
-      def self.new: (untyped name, untyped at, untyped arguments, untyped interval_seconds, untyped missed_policy) -> instance
-                  | (name: untyped, at: untyped, arguments: untyped, interval_seconds: untyped, missed_policy: untyped) -> instance
+      def self.new: (untyped name, untyped operation, untyped at, untyped arguments, untyped interval_seconds, untyped missed_policy) -> instance
+                  | (name: untyped, operation: untyped, at: untyped, arguments: untyped, interval_seconds: untyped, missed_policy: untyped) -> instance
 
-      def self.members: () -> [ :name, :at, :arguments, :interval_seconds, :missed_policy ]
+      def self.members: () -> [ :name, :operation, :at, :arguments, :interval_seconds, :missed_policy ]
 
-      def members: () -> [ :name, :at, :arguments, :interval_seconds, :missed_policy ]
+      def members: () -> [ :name, :operation, :at, :arguments, :interval_seconds, :missed_policy ]
     end
 
     class OutboundMessageIntent < Data
@@ -157,8 +164,30 @@ module SolidObjects
     # @rbs (Symbol | String, **untyped) -> nil
     def commit_action: (Symbol | String, **untyped) -> nil
 
-    # @rbs (at: Time, ?every: Numeric?, ?missed: Symbol | String) -> OperationDispatcher
-    def schedule: (at: Time, ?every: Numeric?, ?missed: Symbol | String) -> OperationDispatcher
+    # A reminder is identified by its name, and without a key that name is the
+    # operation, so one actor holds one alarm per operation. A key gives an actor
+    # an alarm per item it is waiting on, which is what an actor holding a queue
+    # of scheduled work needs; the key is the caller's own identifier for the
+    # item, and scheduling the same key again moves that item's alarm.
+    # @rbs (at: Time, ?every: Numeric?, ?missed: Symbol | String, ?key: (String | Symbol | Integer)?) -> OperationDispatcher
+    def schedule: (at: Time, ?every: Numeric?, ?missed: Symbol | String, ?key: (String | Symbol | Integer)?) -> OperationDispatcher
+
+    # @rbs ((String | Symbol | Integer)?) -> String?
+    def validated_reminder_key: ((String | Symbol | Integer)?) -> String?
+
+    # A keyed name is the operation, a colon, and the key, so an operation
+    # holding a colon of its own would make two different schedules produce one
+    # name: an unkeyed "deliver:item" and a "deliver" keyed "item" would share a
+    # row, and the second would silently take the first one's alarm. Refusing a
+    # colon in the operation keeps unkeyed names free of colons, which leaves
+    # the two kinds of name disjoint and lets a key hold colons of its own.
+    #
+    # The length is checked on the composed name rather than the key alone,
+    # because a long operation and a short key can exceed the column just as
+    # easily as the reverse. Both are refused here rather than at the insert,
+    # once the turn is already doing work.
+    # @rbs (operation: Symbol | String, key: String?) -> String
+    def reminder_name: (operation: Symbol | String, key: String?) -> String
 
     # @rbs (Reference, ?available_at: Time?, ?idempotency_key: String?) -> OperationDispatcher
     def send_to: (Reference, ?available_at: Time?, ?idempotency_key: String?) -> OperationDispatcher

sig/generated/lib/solid_objects/executor.rbs

@@ -46,10 +46,10 @@ module SolidObjects
     def enqueue_effects: (message: Message, instance: Instance, intents: Array[Actor::EffectIntent]) -> Array[Effect]
 
     # 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.
+    # already armed moves it rather than adding a second. Without a key that
+    # name is the operation, so an actor arming a reminder per queued item keeps
+    # only the last; passing schedule a key gives each item its own name and so
+    # its own alarm.
     # 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

test/integration/reminders_test.rb

@@ -31,6 +31,11 @@ def add(wait_until:)
       schedule(at: Time.at(wait_until).utc).deliver
     end
 
+    def add_keyed(item:, wait_until:)
+      self.entries = entries + [ wait_until ]
+      schedule(at: Time.at(wait_until).utc, key: item).deliver
+    end
+
     def arm(name:, wait_until:)
       schedule(at: Time.at(wait_until).utc).public_send(name.to_sym)
     end
@@ -160,6 +165,58 @@ def arm_next
   # 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 key gives each queued item its own reminder" do
+    reference = QueueActor.ref("keyed")
+    first = 1.hour.from_now.to_i
+    second = 2.hours.from_now.to_i
+    reference.async.add_keyed(item: "a", wait_until: first)
+    reference.async.add_keyed(item: "b", wait_until: second)
+    SolidObjects::Worker.new.run_until_idle
+
+    reminders = SolidObjects::Reminder.where(actor_type: "reminder-queue").order(:next_run_at)
+    assert_equal 2, reminders.count, "each key is its own alarm"
+    assert_equal [ first, second ], reminders.map { |reminder| reminder.next_run_at.to_i }
+  end
+
+  test "a keyed reminder still runs the operation it was scheduled with" do
+    reference = QueueActor.ref("keyed-operation")
+    reference.async.add_keyed(item: "a", wait_until: 1.hour.from_now.to_i)
+    SolidObjects::Worker.new.run_until_idle
+
+    reminder = SolidObjects::Reminder.where(actor_type: "reminder-queue").sole
+    assert_equal "deliver", reminder.operation
+    assert_equal "deliver:a", reminder.name
+  end
+
+  test "scheduling the same key again moves that item's reminder" do
+    reference = QueueActor.ref("keyed-move")
+    reference.async.add_keyed(item: "a", wait_until: 1.hour.from_now.to_i)
+    moved = 3.hours.from_now.to_i
+    reference.async.add_keyed(item: "a", wait_until: moved)
+    SolidObjects::Worker.new.run_until_idle
+
+    reminders = SolidObjects::Reminder.where(actor_type: "reminder-queue")
+    assert_equal 1, reminders.count, "the same key is the same alarm"
+    assert_equal moved, reminders.sole.next_run_at.to_i
+  end
+
+  test "a keyed reminder delivers to its actor" do
+    reference = QueueActor.ref("keyed-delivery")
+    reference.async.add_keyed(item: "a", wait_until: 2.seconds.ago.to_i)
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+    scheduler = SolidObjects::ReminderScheduler.new
+
+    assert scheduler.run_once, "the keyed alarm should have come due"
+    worker.run_until_idle
+
+    state = SolidObjects::Instance.find_by!(actor_type: "reminder-queue", actor_id: "keyed-delivery").state
+    assert_empty state.fetch("entries")
+  ensure
+    scheduler&.stop
+    worker&.stop
+  end
+
   test "a second schedule with the same name moves the existing reminder" do
     reference = QueueActor.ref("table")
     first = 1.hour.from_now.to_i
@@ -228,13 +285,15 @@ def arm_next
     )
     intent = SolidObjects::Actor::ReminderIntent.new(
       name: "deliver",
+      operation: "deliver",
       at: 1.hour.from_now,
       arguments: {},
       interval_seconds: nil,
       missed_policy: "latest"
     )
     later = SolidObjects::Actor::ReminderIntent.new(
       name: "deliver",
+      operation: "deliver",
       at: 2.hours.from_now,
       arguments: {},
       interval_seconds: nil,