Class: Yes::Core::CommandHandling::RevisionConflictBackoff

Inherits:

Object

Object
Yes::Core::CommandHandling::RevisionConflictBackoff

show all

Defined in:: lib/yes/core/command_handling/revision_conflict_backoff.rb

Overview

Decides how long an executor waits before retrying a command after a PgEventstore::WrongExpectedRevisionError.

The executors derive the expected revision from the read model's revision column. When the conflicting append came from this service, the read model was updated in the same process, so the very next attempt already sees the new revision and any wait only adds latency. When another service appended to the stream, the column only advances once this service's event listener has projected that event, typically a few hundred milliseconds later. Retrying immediately then burns every attempt inside the same half second and the command surfaces as a 500.

The error carries the stream's actual revision, so for the aggregate's own stream the two cases can be told apart: retry at once while the reloaded read model has caught up, back off while it is still behind. A conflict on an external aggregate's stream (see EventPublisher#verify_external_revisions!) cannot be checked here and always backs off.

Delays follow the same 10 ms doubling schedule as the ConcurrentUpdateError branch, capped per attempt and in total, with jitter so that requests which collided once do not retry in lockstep.

Constant Summary collapse

BASE_DELAY_SECONDS = Returns delay of the first waiting attempt, in seconds. Returns: (Float) — delay of the first waiting attempt, in seconds

0.01

MAX_DELAY_SECONDS = Returns longest delay of a single attempt, in seconds. Returns: (Float) — longest delay of a single attempt, in seconds

1.0

TOTAL_BUDGET_SECONDS = Returns total sleep budget across all retries of one command, in seconds. Returns: (Float) — total sleep budget across all retries of one command, in seconds

2.0

JITTER_FRACTION = Returns fraction by which a delay is randomised in both directions. Returns: (Float) — fraction by which a delay is randomised in both directions

0.25

Class Method Summary collapse

.delay(error:, aggregate_id:, read_model_revision:, attempt:) ⇒ Float
Seconds to wait before retrying, 0.0 to retry immediately.
.schedule(attempt) ⇒ Float
The undisturbed exponential schedule, shared with the ConcurrentUpdateError retries.

Class Method Details

.delay(error:, aggregate_id:, read_model_revision:, attempt:) ⇒ `Float`

Returns seconds to wait before retrying, 0.0 to retry immediately.

Parameters:

error (PgEventstore::WrongExpectedRevisionError) —
the conflict that was raised
aggregate_id (String) —
id of the aggregate the executor works on; used to tell its own stream from an external aggregate's stream
read_model_revision (Integer, nil) —
the revision the read model reports after a reload, or nil when the aggregate has no read model
attempt (Integer) —
the 1-based retry attempt about to be made

Returns:

(Float) —
seconds to wait before retrying, 0.0 to retry immediately

# File 'lib/yes/core/command_handling/revision_conflict_backoff.rb', line 45

def delay(error:, aggregate_id:, read_model_revision:, attempt:)
  return 0.0 unless worth_waiting?(error, aggregate_id, read_model_revision)

  remaining = TOTAL_BUDGET_SECONDS - waited_before(attempt)
  return 0.0 unless remaining.positive?

  jittered([schedule(attempt), remaining].min)
end

.schedule(attempt) ⇒ `Float`

The undisturbed exponential schedule, shared with the ConcurrentUpdateError retries.