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

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.



138
139
140
141
142
143
 
# File 'lib/wurk/api/idempotency.rb', line 138

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 



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

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

.in_progress(request) ⇒ Object 



157
158
159
160
161
162
163
164
165
 
# File 'lib/wurk/api/idempotency.rb', line 157

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

.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.



114
115
116
117
 
# File 'lib/wurk/api/idempotency.rb', line 114

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 



119
120
121
 
# File 'lib/wurk/api/idempotency.rb', line 119

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

.replay(request, stored, fingerprint) ⇒ Object 



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

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.



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

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, and a raise is a request whose outcome nobody knows — both release the key rather than pinning an answer to it.



102
103
104
105
106
107
108
109
 
# File 'lib/wurk/api/idempotency.rb', line 102

def settle(slot, fingerprint)
  response = yield
  response[0] < 300 ? record(slot, fingerprint, response) : release(slot)
  response
rescue StandardError
  release(slot)
  raise
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