Module: Wurk::API::Idempotency

Defined in:
lib/wurk/api/idempotency.rb

Overview

Idempotency-Key replay protection for the produce plane.

A producer whose connection drops mid-POST /jobs cannot tell a lost request from a lost response, so its only safe move is to send it again — and without this, the retry is a second job. The header turns that retry into a replay: the first request's response comes back verbatim and nothing is enqueued twice.

Entirely client-driven. No header means no key, no record and no round trip, so a producer that does not need this pays nothing for it.

Constant Summary collapse

HEADER = 
'HTTP_IDEMPOTENCY_KEY'
KEY_FORMAT =

The same shape a bearer token has to survive: printable ASCII, no spaces. A key that loses a byte in transit would silently address a different record, which is the one failure this module exists to prevent.

/\A[\x21-\x7e]{1,255}\z/
REPLAY_HEADER =

Marks a replayed response, so a client can tell "your job was enqueued" from "your job was enqueued, earlier".

'idempotency-replayed'
PENDING =

A record is <status>\n<body digest>\n<body>, and status 0 means a claim still in flight. Line-oriented rather than JSON because it is read back under contention: a truncated or hand-poisoned value has to degrade to "still in progress", not to a parser exception on the request path.

0
PRE_APPLY_ERRORS =

Errors that prove the write they interrupted never applied: refused while dialing, a starved checkout, or a reply in which Redis itself rejected the command.

[
  ::RedisClient::CannotConnectError, ::RedisClient::FailoverError,
  ::RedisClient::CommandError, ::ConnectionPool::TimeoutError
].freeze

Class Method Summary collapse

Class Method Details

.around(request, raw_body, config, &handler) ⇒ Object

Wraps a state-changing handler. Without the header this is a bare call through; with it, exactly one of the requests sharing a key runs the handler and the rest replay its answer.

Parameters:

  • raw_body (String) —

    the bytes the response is pinned to, so the same key sent with a different body is a client bug, not a replay.



49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/wurk/api/idempotency.rb', line 49

def around(request, raw_body, config, &handler)
  key = presented(request)
  return handler.call unless key

  slot = slot_for(request, key)
  fingerprint = ::Digest::SHA256.hexdigest(raw_body)
  stored = claim(slot, fingerprint, config.api_idempotency_ttl)
  return replay(request, stored, fingerprint) if stored

  settle(slot, fingerprint, &handler)
end

.claim(slot, fingerprint, ttl) ⇒ String?

One round trip: reserve the key if it is free, and read what is there either way. Deliberately not a MULTI — the SET already picked the winner, and the GET only has to explain the loss.

Returns:

  • (String, nil) —

    nil when this request owns the key; otherwise the record the owner left, or is still to leave. An empty string when the owner's record expired in between, which reads as "in flight" and is the safe answer: it never enqueues a second job.



88
89
90
91
92
93
94
95
96
 
# File 'lib/wurk/api/idempotency.rb', line 88

def claim(slot, fingerprint, ttl)
  reserved, existing = ::Wurk.redis do |conn|
    conn.pipelined do |pipe|
      pipe.call('SET', slot, encode(PENDING, fingerprint, ''), 'NX', 'EX', ttl)
      pipe.call('GET', slot)
    end
  end
  reserved ? nil : existing.to_s
end

.decode(stored) ⇒ Array(Integer, String, String)?

Returns nil for a record that is gone or unreadable — the caller treats both as still in flight.

Returns:

  • (Array(Integer, String, String), nil) —

    nil for a record that is gone or unreadable — the caller treats both as still in flight.



161
162
163
164
165
166
 
# File 'lib/wurk/api/idempotency.rb', line 161

def decode(stored)
  status, digest, body = stored.split("\n", 3)
  return nil if digest.nil? || digest.empty?

  [status.to_i, digest, body.to_s]
end

.encode(status, digest, body) ⇒ Object 



157
 
# File 'lib/wurk/api/idempotency.rb', line 157

def encode(status, digest, body) = "#{status}\n#{digest}\n#{body}"

.in_progress(request) ⇒ Object 



180
181
182
183
184
185
186
187
188
 
# File 'lib/wurk/api/idempotency.rb', line 180

def in_progress(request)
  Problem.render(
    Problem::REQUEST_IN_PROGRESS,
    status: 409,
    detail: 'A request with this Idempotency-Key is still in flight.',
    instance: request.path,
    headers: { 'retry-after' => '1' }
  )
end

.outcome_unknown?(error) ⇒ Boolean

Returns:

  • (Boolean) 


126
127
128
129
130
131
132
 
# File 'lib/wurk/api/idempotency.rb', line 126

def outcome_unknown?(error)
  case Thread.current[::Wurk::Client::WRITE_STATE_KEY]
  when :applied then true
  when :attempted then PRE_APPLY_ERRORS.none? { |klass| error.is_a?(klass) }
  else false
  end
end

.presented(request) ⇒ Object

Raises:



61
62
63
64
65
66
67
 
# File 'lib/wurk/api/idempotency.rb', line 61

def presented(request)
  key = request.get_header(HEADER)
  return nil if key.nil?
  return key if KEY_FORMAT.match?(key)

  raise Validation::Invalid, 'Idempotency-Key must be 1-255 printable ASCII characters with no spaces.'
end

.record(slot, fingerprint, response) ⇒ Object

KEEPTTL so the window is the first request's, not this write's, and XX so a claim whose TTL lapsed mid-flight is not resurrected as a record with no expiry at all.



137
138
139
140
 
# File 'lib/wurk/api/idempotency.rb', line 137

def record(slot, fingerprint, response)
  status, _headers, body = response
  ::Wurk.redis { |conn| conn.call('SET', slot, encode(status, fingerprint, body.join), 'XX', 'KEEPTTL') }
end

.release(slot) ⇒ Object 



142
143
144
 
# File 'lib/wurk/api/idempotency.rb', line 142

def release(slot)
  ::Wurk.redis { |conn| conn.call('DEL', slot) }
end

.replay(request, stored, fingerprint) ⇒ Object 



146
147
148
149
150
151
152
153
154
155
 
# File 'lib/wurk/api/idempotency.rb', line 146

def replay(request, stored, fingerprint)
  record = decode(stored)
  return in_progress(request) unless record

  status, digest, body = record
  return reused(request) unless digest == fingerprint
  return in_progress(request) if status == PENDING

  [status, Response::HEADERS.merge(REPLAY_HEADER => 'true'), [body]]
end

.reused(request) ⇒ Object

409 rather than 400: the request is well-formed, and what it collides with is a fact about the server's state that the client can resolve by rotating the key.



171
172
173
174
175
176
177
178
 
# File 'lib/wurk/api/idempotency.rb', line 171

def reused(request)
  Problem.render(
    Problem::IDEMPOTENCY_KEY_REUSED,
    status: 409,
    detail: 'This Idempotency-Key was already used for a different request body.',
    instance: request.path
  )
end

.settle(slot, fingerprint) ⇒ Object

Only a success is worth replaying. A rejected body is a request the client should be able to correct and send again under the same key, so it releases. A raise releases only when nothing can have reached Redis — before any write went out, or on an error that proves the one in flight never applied. Anything else (a lost reply, a bulk request that died after an earlier slice landed) is an outcome nobody knows: the key stays pending until its TTL, so the client's retry is told "in flight" instead of enqueueing a second copy.



114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/wurk/api/idempotency.rb', line 114

def settle(slot, fingerprint)
  Thread.current[::Wurk::Client::WRITE_STATE_KEY] = :clean
  response = yield
  response[0] < 300 ? record(slot, fingerprint, response) : release(slot)
  response
rescue StandardError => e
  release(slot) unless outcome_unknown?(e)
  raise
ensure
  Thread.current[::Wurk::Client::WRITE_STATE_KEY] = nil
end

.slot_for(request, key) ⇒ Object

Scoped to the credential and to the route. The key is a string the client chose: two producers that both sent 1 must not see each other's jids, and the same key on /jobs and /jobs/bulk addresses two different requests. Hashed rather than concatenated so the client's own string never reaches Redis in the clear and every slot is one length.



74
75
76
77
78
 
# File 'lib/wurk/api/idempotency.rb', line 74

def slot_for(request, key)
  ::Wurk::Keys.idempotency(
    ::Digest::SHA256.hexdigest("#{request.principal.id}\n#{request.path_info}\n#{key}")
  )
end