Merge pull request #315 from amazon-mq/improve-upload-retry-strategy-212

Add exponential backoff and stalled-upload observability to the retry path (#212)

Author: the-mikedavis

docs/failure-modes.md

@@ -27,6 +27,20 @@ For the concepts behind these scenarios, see [concepts.md](./concepts.md). For t
 
 ---
 
+## Upload retry strategy
+
+A failed fragment upload is always retried; the fragment is never abandoned, because dropping it would advance the manifest over a range that is not durable in S3 and leave a silent hole (issue #206). How the retry is paced depends on the error class.
+
+**Transient errors** (throttling, 5xx, timeouts, connection errors) are expected to clear quickly. The first retry is immediate, preserving responsiveness for a one-off blip. Successive consecutive failures of the same fragment back off exponentially from `stream_s3.task_retry_delay_constant` (10ms) by a factor of `stream_s3.task_retry_delay_exponent` (2), capped at `stream_s3.task_retry_delay_max_ms` (5s).
+
+**Non-transient errors** (a confirmed checksum mismatch, an unexpected 4xx) are unlikely to clear on a tight retry, so the pipeline stalls at the failed offset and retries with a backoff starting at `upload_retry_delay_ms` (1000ms), growing to `upload_retry_delay_max_ms` (30s). Local-tier cleanup also stalls at that offset, so the only durable copy is retained until the upload succeeds.
+
+Both profiles apply equal jitter (the delay is uniformly distributed in `[delay/2, delay]`) so that many streams stalled by a shared incident do not retry against S3 in lockstep. The per-fragment attempt counter resets once the fragment is durable.
+
+**Detection.** `rate(rabbitmq_stream_s3_transfer_retries[5m])` rising means uploads are not succeeding on the first try. `rate(rabbitmq_stream_s3_nontransient_transfer_retries[5m]) > 0` or `rabbitmq_stream_s3_upload_stalled_offset > 0` means a fragment is failing with a confirmed-fatal error and the pipeline is wedged at that offset; the accompanying WARNING log names the error and offset.
+
+---
+
 ## Leader election
 
 **Trigger.** Writer node fails or is shut down. A new writer is elected.

docs/operations.md

@@ -163,12 +163,15 @@ Owned by `rabbitmq_stream_s3_replica_reader`. Each writer-node replica reader ow
 |-------------------------------------|---------|-------------------------------------------------------------------|
 | `rabbitmq_stream_s3_transfers_completed`               | counter | Fragment uploads that succeeded                                   |
 | `rabbitmq_stream_s3_transfers_failed`                  | counter | Fragment uploads that failed for any reason                       |
+| `rabbitmq_stream_s3_transfer_retries`                  | counter | Fragment upload retries scheduled (transient and non-transient errors) |
+| `rabbitmq_stream_s3_nontransient_transfer_retries`     | counter | Fragment upload retries for a confirmed-but-non-transient error (a checksum mismatch, an unexpected 4xx); the pipeline stalls at that offset until the fragment is durable |
 | `rabbitmq_stream_s3_bytes_transferred`                 | counter | Total payload bytes uploaded                                      |
 | `rabbitmq_stream_s3_groups_created`                    | counter | Group manifest objects uploaded                                   |
 | `rabbitmq_stream_s3_kilo_groups_created`               | counter | Kilo-group manifest objects uploaded                              |
 | `rabbitmq_stream_s3_mega_groups_created`               | counter | Mega-group manifest objects uploaded                              |
 | `rabbitmq_stream_s3_roots_created`                     | counter | Root manifest objects uploaded (one per successful persist)       |
 | `rabbitmq_stream_s3_transfers_in_flight`               | gauge   | Fragments cut and submitted that have not yet completed           |
+| `rabbitmq_stream_s3_upload_stalled_offset`             | gauge   | Offset at which the pipeline is stalled on a non-transient error, or 0 when not stalled |
 
 #### Pipeline (drain to persist)
 
@@ -372,6 +375,7 @@ The numbers below are starting points. Tune for your traffic patterns.
 - **Persistent transfer failures.** `rate(rabbitmq_stream_s3_transfers_failed[5m]) > 0` for more than ten minutes. Sustained failures usually mean S3 connectivity, throttling, or credentials problems.
 - **Persist conflicts.** `rate(rabbitmq_stream_s3_persist_conflicts[5m]) > 0`. A non-zero rate indicates competing writers, likely from a partition or a struggling leader election.
 - **Stuck uploads.** `rabbitmq_stream_s3_transfers_in_flight` non-zero and `rate(rabbitmq_stream_s3_transfers_completed[5m]) == 0` together. Drain loop is producing fragments but the governor is not flushing them.
+- **Stalled pipeline (non-transient error).** `rabbitmq_stream_s3_upload_stalled_offset > 0`, or `rate(rabbitmq_stream_s3_nontransient_transfer_retries[5m]) > 0`. A fragment is failing with a confirmed-fatal error (for example a persistent checksum mismatch or an unexpected 4xx) and the pipeline is stuck at that offset, retrying with a backoff. Inspect the accompanying WARNING log for the error and offset.
 - **Local log ahead recoveries.** `rate(rabbitmq_stream_s3_local_log_ahead_recoveries[1h]) > 0` for any stream. Indicates local retention is racing ahead of uploads. See [failure-modes.md](./failure-modes.md).
 - **Khepri conflicts.** `rate(rabbitmq_stream_s3_put_conflicts[5m]) > 0` matches `rabbitmq_stream_s3_persist_conflicts` above and surfaces the same problem from the metadata store side.
 - **S3 errors.** `rate(rabbitmq_stream_s3_response_500[5m]) > 0` or `rate(rabbitmq_stream_s3_response_503[5m]) > 0`. The plugin retries automatically; sustained high rates indicate an S3-side incident.

src/rabbitmq_stream_s3_config.erl

@@ -39,6 +39,7 @@ lives here. Callers use these functions instead of calling
     task_retry_delay_max_ms/0,
     task_retry_delay_constant/0,
     task_retry_delay_exponent/0,
+    upload_retry_delay_max_ms/0,
     verbose_logging/0,
     segment_upload_timeout/0,
     upload_retry_delay_ms/0,
@@ -193,13 +194,22 @@ verbose_logging() ->
 segment_upload_timeout() ->
     application:get_env(?APP, segment_upload_timeout, 45_000).
 
-%% Delay before retrying a fragment upload that failed with a non-transient
+%% Base delay before retrying a fragment upload that failed with a non-transient
 %% error. The upload pipeline stalls at the failed offset until the retry
-%% succeeds, so this bounds how often a persistently failing upload is retried.
+%% succeeds. Successive non-transient retries back off exponentially from this
+%% base (using task_retry_delay_exponent) up to upload_retry_delay_max_ms, so a
+%% persistently failing upload is not retried tightly.
 -spec upload_retry_delay_ms() -> non_neg_integer().
 upload_retry_delay_ms() ->
     application:get_env(?APP, upload_retry_delay_ms, 1000).
 
+%% Ceiling for the non-transient upload-retry backoff. A confirmed-fatal error
+%% (a checksum mismatch, an unexpected 4xx) is unlikely to clear on a tight
+%% retry, so the backoff is allowed to grow well beyond the transient ceiling.
+-spec upload_retry_delay_max_ms() -> non_neg_integer().
+upload_retry_delay_max_ms() ->
+    application:get_env(?APP, upload_retry_delay_max_ms, 30_000).
+
 %% Deadline for a submitted fragment transfer to report a result back to the
 %% replica reader. The reader submits each transfer to the per-node governor
 %% and waits for a `{transfer_result, Ref, _}` message. That message can fail
@@ -280,6 +290,7 @@ defaults_test_() ->
         ?_assertEqual(false, verbose_logging()),
         ?_assertEqual(45_000, segment_upload_timeout()),
         ?_assertEqual(1000, upload_retry_delay_ms()),
+        ?_assertEqual(30_000, upload_retry_delay_max_ms()),
         ?_assertEqual(180_000, transfer_deadline_ms()),
         ?_assertEqual(60_000, retention_task_timeout()),
         ?_assertEqual(5000, tick_timeout_milliseconds()),