Class: Insika::OutboxStore

Inherits:
Object
  • Object
show all
Defined in:
lib/insika/outbox_store.rb

Overview

Durable record of one pending OUTBOUND reply (RFC-0011 §6.5). A Shape B channel answers out of band — the turn ends and the reply is POSTed to the platform (or, for a relay, to the consumer's own callback) afterwards — so "the answer exists but the recipient is not on this connection" is exactly the problem RFC-0010 Phase 2 already solved for async delegation. This store is DelegationStore's shape with a different recipient, deliberately: a second invention here would be a second thing to get wrong.

Lifecycle (never backwards):

pending     the turn committed a reply; nobody has claimed it yet. This is
          the ONLY status the boot sweep re-drives.
delivering  claimed. The claim happens BEFORE the HTTP call, so a crash
          here loses the delivery rather than duplicating it —
          at-most-once, the same honest scope as the delegation path.
delivered   the recipient answered 2xx.
failed      the bounded retry ran out. Terminal, and NOT re-driven at boot:
          a third party that refused N times is an operator problem, not
          something to replay forever.

failed and delivering are a deliberate widening of the RFC's two-status sketch (pending -> delivered): without them a crashed claim and an exhausted retry are indistinguishable from a fresh record, and the sweep would redeliver both.

Normalizes symbol->string on WRITE (the backend only round-trips JSON types), like every other domain store.

Defined Under Namespace

Classes: Delivery

Constant Summary collapse

SCOPE = 
"outbox"
KEY_PREFIX = 
"outbox:"
STATUSES = 
%i[pending delivering delivered failed].freeze

Instance Method Summary collapse

Constructor Details

#initialize(store:) ⇒ OutboxStore

Returns a new instance of OutboxStore.



44
45
46
 
# File 'lib/insika/outbox_store.rb', line 44

def initialize(store:)
  @store = store
end

Instance Method Details

#claim(id) ⇒ Object

pending -> delivering, ATOMICALLY — across processes, not just fibers: the read-check-write rides Store#transaction, so two workers claiming the same record serialize on the backend's lock and only one sees :pending. Returns true only for the caller that won the transition; that caller (and only it) makes the HTTP call, so delivery is at-most-once even if the terminal hook and the boot sweep both fire.



88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/insika/outbox_store.rb', line 88

def claim(id)
  @store.transaction do
    record = @store.get(SCOPE, key_for(id))
    next false unless record && record["status"] == "pending"

    record["status"] = "delivering"
    record["updated_at"] = timestamp
    @store.set(SCOPE, key_for(id), record)
    true
  end
end

#create(channel:, to:, task_id:, session_id:, payload:, id: SecureRandom.uuid) ⇒ Object

-> Delivery (:pending). payload is the body the channel will send; it is DATA (string keys, JSON types) and the store never interprets it.



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/insika/outbox_store.rb', line 50

def create(channel:, to:, task_id:, session_id:, payload:, id: SecureRandom.uuid)
  record = {
    "id" => id.to_s,
    "channel" => channel.to_s,
    "to" => to.to_s,
    "task_id" => task_id&.to_s,
    "session_id" => session_id&.to_s,
    "payload" => payload,
    "status" => "pending",
    "attempts" => 0,
    "last_error" => nil,
    "created_at" => timestamp,
    "updated_at" => timestamp
  }
  @store.set(SCOPE, key_for(id), record)
  to_delivery(record)
end

#find(id) ⇒ Object

-> Delivery | nil



69
70
71
72
 
# File 'lib/insika/outbox_store.rb', line 69

def find(id)
  record = @store.get(SCOPE, key_for(id))
  record && to_delivery(record)
end

#mark_delivered(id) ⇒ Object

-> Delivery (:delivered). Idempotent: an already-delivered record is returned unchanged rather than re-marked.



112
113
114
115
116
117
118
119
120
 
# File 'lib/insika/outbox_store.rb', line 112

def mark_delivered(id)
  record = fetch!(id)
  return to_delivery(record) if record["status"] == "delivered"

  record["status"] = "delivered"
  record["attempts"] = record["attempts"].to_i + 1
  record["last_error"] = nil
  touch(id, record)
end

#mark_failed(id, error: nil) ⇒ Object

-> Delivery (:failed). The retry budget is spent; nothing re-drives this.



123
124
125
126
127
128
 
# File 'lib/insika/outbox_store.rb', line 123

def mark_failed(id, error: nil)
  record = fetch!(id)
  record["status"] = "failed"
  record["last_error"] = error&.to_s if error
  touch(id, record)
end

#pending ⇒ Object

-> [Delivery] still waiting for a first claim (boot sweep). Deliberately NOT delivering: that one was claimed by a process that then died, and whether its POST landed is unknowable — replaying it is the duplicate the claim exists to prevent.



78
79
80
 
# File 'lib/insika/outbox_store.rb', line 78

def pending
  scan.select { |d| d.status == :pending }
end

#record_attempt(id, error: nil) ⇒ Object

One attempt happened and did not succeed. Keeps the record :delivering (the in-process retry loop owns it) and writes down why, so an operator reading the store sees the third party's answer and not just a counter.



103
104
105
106
107
108
 
# File 'lib/insika/outbox_store.rb', line 103

def record_attempt(id, error: nil)
  record = fetch!(id)
  record["attempts"] = record["attempts"].to_i + 1
  record["last_error"] = error&.to_s
  touch(id, record)
end