fix(log-store): keep the enqueued backlog on permanent object store errors (#9558)

* fix(log-store): keep the enqueued backlog on permanent object store errors

On object storage the WAL's object store sits behind OpenDAL's retry
layer, which marks every error it returns as persistent. The enqueued
mode classified create failures by that status, so a 403 or a missing
bucket was repeated every 100 ms on every create in flight, logged
nothing, and made appends hang once the sealed batches were full.

Object store errors are now classified by kind: denied access, an
invalid configuration (how OpenDAL reports a missing S3 bucket) and a
write to a missing location are non-retriable. In the enqueued mode such
an error makes the store failing instead of poisoning it: the
acknowledged backlog is kept under its sequences and bytes, its lowest
pending batch is attempted alone with a delay that doubles from 100 ms
to 30 s, appends are rejected at once with a retryable error that
carries the last object store error, and the first create that succeeds
ends the state. Stop uploads the backlog at once and reports a failure
as before. The state is logged on entry, at most once a minute while it
lasts and on exit, and exported as the `failing` gauge and the
`rejected_appends_total` counter. Conflicts, encoding and catalog errors
still poison the store; the durable mode is unchanged.

An append held back because the sealed batches are full is now counted
and timed by the stall metrics in both modes.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* docs(rfc): describe how the enqueued object store WAL handles refused creates

Describe the enqueued failure contract as the store now implements it:
non-retriable object store errors are told apart by kind because
OpenDAL's retry layer under the store marks every error persistent, a
failing store keeps its backlog and attempts it with backoff while it
rejects appends, and stop uploads the backlog at once. Add the failure
matrix row, the failing gauge and the rejected appends counter, the
sealed-batch admission limit and the stall metrics that now count it,
and the fault matrix case counts.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* fix(log-store): time the failing attempts where they start and fix two statements

The test of the attempts of a failing store now records when each
refused write starts and checks the gaps between them, instead of
sampling a counter after a sleep of the test task. The delay comment and
the RFC describe 30 seconds as the longest delay between two attempts,
not as a bound on when appends resume, and the RFC states that the next
append is held once seven sealed batches wait.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* test(log-store): drop the wait_until helper that main now defines

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* refactor(log-store): name errors that need an operator and trim the RFC contract

Rename is_non_retriable to requires_intervention: such a create is still
repeated, only slowly, until someone fixes the credentials, configuration
or bucket. Name the backoff test after what it asserts, cut the test of
the lowest pending batch to two failed rounds since the cadence is
covered by the backoff test, and shorten the RFC failure contract to
the contract and its rationale, leaving error mappings, request counts
and logging cadence to the code comments.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* docs(log-store): tie the slow repeat to errors requiring intervention

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

docs/rfcs/2026-09-06-object-store-wal.md

@@ -121,13 +121,17 @@ The store then selects the history it replays from the headers. An object's link
 
 Before it admits any append, the store writes a start object: an object without segments at the next sequence, which is above every listed object, orphans included, linked to the tip, or to nothing on an empty prefix. The instance's epoch is the sequence of its start object plus one, and every object the instance writes carries it. The conditional create assigns each sequence to exactly one object, so no two instances share an epoch; and the start sequence is above every object present at open, including the start objects of every earlier instance, so the epoch of every later instance is greater than that of any object an earlier instance ever writes, including one whose create lands after that instance stopped or crashed, and such an object never becomes the tip. Only once the start object is created does the store accept appends, and the first batch links to it. If the sequence turns out to be taken, the store reads the object there: an object of an earlier epoch is stale, and the start object moves to the next sequence, taking the epoch that sequence gives; an object of the same or a later epoch means another writer, and construction fails with a conflict error that is not retried; an object with identical content, which another open of the same prefix has just written, fails construction with a retryable error, so that the next attempt lists it and starts above it. A create whose outcome is unknown fails construction with a retryable error too: an object it may have left is listed by the next attempt, which therefore starts above it. Two assumptions remain. Listing the prefix is strongly consistent, so an open sees every start object that was durable before it listed; this holds for S3, GCS, Azure Blob Storage and MinIO, and an S3-compatible store without it is not supported. And one process at a time opens a prefix: two processes opening it concurrently either contend for the same start sequence, where the conditional create lets one through and fails the other's construction, or, if one of them listed a late object the other did not, start under different sequences and epochs and both proceed until their sequences meet, where the writer that meets the other's object follows the conflict rules under *Writing* and the *Failure matrix*: in the `durable` mode an object of an earlier epoch fails its batch and one of a later epoch poisons it, and in the `enqueued` mode any conflict poisons it, so which writer is poisoned depends on who observes the collision, not on the epochs alone; the store does not fence concurrent writers beyond that, and in cluster mode the generation in the key prefix does.
 
-**Writing.** One background actor per store admits appended entries into a single open batch and assigns entry ids as described under *Entry ids* above. The batch is sealed when its estimated size reaches `max_batch_bytes` or when `flush_interval` elapses with entries in it; a timer tick on an empty batch creates nothing. Sealing encodes the batch under the next sequence number and hands it to a pipelined uploader: the actor keeps admitting entries into the next batch while uploads are in flight, and up to four conditional creates run at a time, started in sequence order. Every sealed batch is encoded with the store's epoch and a link to its pipeline predecessor: the last batch sealed before it that is not indexed yet, or, when there is none, the last indexed object, which after opening is the start object. A link to the last indexed object while an earlier batch is still uploading would let a later chain skip that batch once it is acknowledged. Batches are indexed in sequence order in both acknowledgement modes: a created object is not indexed until every earlier batch is. The rest of this paragraph is `durable` mode behaviour; the `enqueued` mode, whose appends have already returned, is described under *Acknowledgement*. In the `durable` mode a batch is acknowledged when it is indexed, so a batch whose object is durable is not acknowledged until every earlier batch is durable too, and the acknowledged prefix of a region's history never has a missing predecessor. When a create fails transiently, every batch that is not indexed fails at once, including batches whose objects were already created and batches whose creates are still in flight. Their sequence numbers are never handed out again, the next batch links to the last indexed object, and the store keeps serving. A create still in flight completes later and finds no batch to acknowledge, so whatever it stored is an orphan. The next object acknowledged links past every failed batch, and from then on recovery replays none of them, whether their objects exist, land later or never do. A create that meets a different object reads that object's header: an object of an earlier epoch is stale and can never be the tip, so the batch fails as for a transient error and the store keeps serving; an object of the same or a later epoch belongs to another writer on the prefix, and the store poisons itself.
+**Writing.** One background actor per store admits appended entries into a single open batch and assigns entry ids as described under *Entry ids* above. The batch is sealed when its estimated size reaches `max_batch_bytes` or when `flush_interval` elapses with entries in it; a timer tick on an empty batch creates nothing. Sealing encodes the batch under the next sequence number and hands it to a pipelined uploader: the actor keeps admitting entries into the next batch while uploads are in flight, and up to four conditional creates run at a time, started in sequence order. At most eight sealed batches wait to be indexed: once seven do, which leaves no room for the two one admission can seal, the next append is held back in both acknowledgement modes until a batch is indexed, and later appends wait behind it. Every sealed batch is encoded with the store's epoch and a link to its pipeline predecessor: the last batch sealed before it that is not indexed yet, or, when there is none, the last indexed object, which after opening is the start object. A link to the last indexed object while an earlier batch is still uploading would let a later chain skip that batch once it is acknowledged. Batches are indexed in sequence order in both acknowledgement modes: a created object is not indexed until every earlier batch is. The rest of this paragraph is `durable` mode behaviour; the `enqueued` mode, whose appends have already returned, is described under *Acknowledgement*. In the `durable` mode a batch is acknowledged when it is indexed, so a batch whose object is durable is not acknowledged until every earlier batch is durable too, and the acknowledged prefix of a region's history never has a missing predecessor. When a create fails transiently, every batch that is not indexed fails at once, including batches whose objects were already created and batches whose creates are still in flight. Their sequence numbers are never handed out again, the next batch links to the last indexed object, and the store keeps serving. A create still in flight completes later and finds no batch to acknowledge, so whatever it stored is an orphan. The next object acknowledged links past every failed batch, and from then on recovery replays none of them, whether their objects exist, land later or never do. A create that meets a different object reads that object's header: an object of an earlier epoch is stale and can never be the tip, so the batch fails as for a transient error and the store keeps serving; an object of the same or a later epoch belongs to another writer on the prefix, and the store poisons itself.
 
 The guarantee is therefore about acknowledged history. An acknowledged entry is always replayed; an entry whose append failed is replayed only when nothing was acknowledged after it, the same outcome as a crash between an object's creation and its acknowledgement, which the engine already tolerates. The distinction matters because Mito does not advance a region's committed sequence when an append fails and reuses those row sequences in its next write, so replaying a failed batch next to a later acknowledged one could let the unacknowledged rows win deduplication.
 
 **Acknowledgement.** `ack_mode` selects when `append_batch` returns. `durable`, the default, returns only after the object holding the entries is durable and indexed, so callers never see an entry id for an entry that is not durable; a write-ahead log is expected to mean durability on return, which is why this stays the default. `enqueued` returns as soon as the entries are admitted, with their entry ids already assigned, and creates the object in the background. In that mode the recovery point objective is the unpersisted backlog. `max_unpersisted_bytes` and `max_unpersisted_age` are admission thresholds: when the open batch plus the sealed batches that are not durable yet reach either, new appends are held back in arrival order until an upload completes, and the store seals the open batch if no create is in flight so that there is an upload to wait for; nothing is dropped and the backlog cannot grow without limit. They are not a hard bound on what a crash can lose: during an outage the writes already acknowledged stay in memory for as long as the outage lasts, however old they get, and a crash in that window loses them.
 
-The failure contract of `enqueued` differs from the durable one because no caller remains to retry. A transient create failure is repeated by the store after a short delay instead of failing the batch. A permanent failure, meaning a conflicting object or an encoding or catalog error, poisons the store. Once `stop` has begun a transient failure is no longer repeated: the backlog is dropped and `stop` reports the loss as its error, since those entries were acknowledged but never became durable. Otherwise `stop` seals and uploads the remaining backlog before it returns.
+The failure contract of `enqueued` differs from the durable one because no caller remains to retry: a create the object store did not confirm is repeated under the same sequence with the same bytes instead of failing the batch. A transient failure is repeated after 100 ms while appends are still admitted. A failure that requires intervention, such as denied access or a missing bucket, makes the store failing. The store tells the two apart by the kind of the object store error, not by its status, because OpenDAL's retry layer under the store marks every error it returns as persistent.
+
+A failing store keeps its acknowledged backlog instead of poisoning itself, since poisoning would drop those entries and need a restart for a cause that can be fixed in place. It attempts the lowest pending batch alone, with exponential backoff, and rejects every new append at once with a retryable error that carries the last object store error, so no append is admitted or waits behind the backlog. A durability wait for an entry in the backlog keeps waiting. The first create that succeeds ends the state: the backlog is uploaded and appends are admitted again.
+
+A conflicting object or an encoding or catalog error poisons the store. Once `stop` has begun a failed create is no longer repeated: the store attempts its whole backlog at once, failing or not, and any object store failure drops the backlog and `stop` reports the loss as its error, since those entries were acknowledged but never became durable. Otherwise `stop` seals and uploads the remaining backlog before it returns.
 
 The mode also needs a barrier that the durable mode gets for free. An entry id that Mito has been handed may not yet be durable, and a flush that persisted such an id as the manifest's `flushed_entry_id` would, after a crash before the object was created, leave a watermark that refers to an object which never existed, so that the sequence is reused on restart and later entries land below the watermark and are skipped by replay. The `LogStore` trait therefore has `wait_durable(provider, entry_id)`, with a default that returns at once, so that every existing provider keeps its own acknowledgement and persistence semantics, whatever they are; the durability guarantee this section describes belongs to the object store provider alone. The object store WAL answers it once the region is durable and indexed through `entry_id`; an id the store never handed out has nothing to wait for. Mito waits before it publishes a manifest watermark: a flush waits after its SSTs are written and before the manifest edit, a full truncate waits before it records the truncated entry id as the replay frontier, and discarding unflushed data waits before it records the discarded entry id as flushed. The wait in a flush observes the flush's cancellation, so a cancelled flush does not hold up the drop or truncate queued behind it; a full truncate and a discard of unflushed data still run their own wait before publishing their frontier. After a backlog was lost, a durability wait for an entry that is not durable fails rather than certifying it, whether the entry was in that backlog or not. In `enqueued` mode `obsolete` also clamps the watermark to the region's durable entry id, since an entry that is not durable yet is replayed after a crash and hiding it would skip it. A manifest watermark can then only ever name a durable entry, and the restart allocation rule under *Entry ids* keeps every new id above it.
 
@@ -139,7 +143,8 @@ The mode also needs a barrier that the durable mode gets for free. An entry id t
 | transient I/O error, `durable` mode | consumed and never reused, nor are the sequences of the other failed batches | this and every batch that is not indexed fail, including those whose objects were created; a create still in flight completes as an orphan | healthy: the next batch links to the last indexed object |
 | an object of an earlier epoch holds the sequence, `durable` mode | consumed and never reused | as for a transient I/O error | healthy |
 | transient I/O error, `enqueued` mode | unchanged | already acknowledged | healthy: the create is repeated after a short delay |
-| transient I/O error, `enqueued` mode after `stop` began | unchanged | already acknowledged | the backlog is dropped and `stop` reports the loss |
+| I/O error that requires intervention, `enqueued` mode | unchanged | already acknowledged; later appends are rejected with a retryable error until a create succeeds | failing: the lowest pending create is repeated alone with backoff; the first create that succeeds ends it |
+| any I/O error, `enqueued` mode after `stop` began | unchanged | already acknowledged | the backlog is dropped and `stop` reports the loss |
 | an object of the same or a later epoch holds the sequence, or any conflicting object in the `enqueued` mode | unchanged | every unacknowledged waiter fails, including those of batches whose objects were created but not yet indexed; an `enqueued` acknowledgement already returned cannot be revoked | poisoned |
 | encoding or catalog error | unchanged | every unacknowledged waiter fails, including those of batches whose objects were created but not yet indexed; an `enqueued` acknowledgement already returned cannot be revoked | poisoned |
 | create succeeds at the last representable sequence | cannot advance | acknowledged | poisoned: no later batch can be allocated a sequence, every later append fails |
@@ -148,11 +153,11 @@ The mode also needs a barrier that the durable mode gets for free. An entry id t
 
 A store cannot be constructed on a prefix whose largest object already carries the maximum sequence. A poisoned store fails every `LogStore` operation except `stop` with the same terminal error; in particular `obsolete` can no longer move a watermark. In the `durable` mode transient errors are not retried inside the store, but a failed create is reconciled once: the store immediately reads the key back, accepts the batch if the object exists with identical bytes, treats it as a conflict if the bytes differ, and reports the transient error only when the key is absent or that read fails too. So a lost response alone does not fail the append. When the append does fail, neither the store nor Mito keeps the batch: it surfaces through Mito to the caller of the write, and a retry is a new write by the caller under a new sequence. If the create had in fact succeeded and both its response and the read-back were lost, or its request lands only later, the object is an orphan as soon as a later object is acknowledged. Retrying the batch itself inside the store in the `durable` mode would only hide the latency and would hold writes up for as long as an outage lasts; the `enqueued` mode repeats the create with the same bytes, so a create that succeeded without a response is accepted as an identical retry.
 
-**Stopping.** `stop` is idempotent and awaits the actor. Once stop has begun, queued appends are not admitted, no timer or size triggered seal starts, and appends after stop, including empty ones, fail with the stopped error. In the `durable` mode the open batch and every sealed batch whose create has not started are dropped: their waiters receive the stopped error. A create that is already in flight runs to completion; if it succeeds while its batch is still pending, the batch is indexed in order and acknowledged. A transient failure after stop began fails every batch that is not indexed, as it does while the store runs, and those batches receive the stopped error; a create of such a batch that completes later is an orphan and acknowledges nothing. A create that finds a conflicting object of the same or a later epoch after stop began poisons the store as it would while the store runs, but the waiters of the batches that are not indexed receive the stopped error, the conflict is kept only as the terminal error, and `stop` returns successfully once the creates in flight have completed. In the `enqueued` mode the held-back appends receive the stopped error, the open batch is sealed and the whole backlog is uploaded before `stop` returns; a transient failure in that phase drops the backlog and `stop` returns the error, as described under *Acknowledgement*. A durability wait still pending when the actor exits receives the stopped error, and a delete already in flight runs to completion before `stop` returns.
+**Stopping.** `stop` is idempotent and awaits the actor. Once stop has begun, queued appends are not admitted, no timer or size triggered seal starts, and appends after stop, including empty ones, fail with the stopped error. In the `durable` mode the open batch and every sealed batch whose create has not started are dropped: their waiters receive the stopped error. A create that is already in flight runs to completion; if it succeeds while its batch is still pending, the batch is indexed in order and acknowledged. A transient failure after stop began fails every batch that is not indexed, as it does while the store runs, and those batches receive the stopped error; a create of such a batch that completes later is an orphan and acknowledges nothing. A create that finds a conflicting object of the same or a later epoch after stop began poisons the store as it would while the store runs, but the waiters of the batches that are not indexed receive the stopped error, the conflict is kept only as the terminal error, and `stop` returns successfully once the creates in flight have completed. In the `enqueued` mode the held-back appends receive the stopped error, the open batch is sealed and the whole backlog is uploaded before `stop` returns, by a failing store too; any object store failure in that phase drops the backlog and `stop` returns the error, as described under *Acknowledgement*. A durability wait still pending when the actor exits receives the stopped error, and a delete already in flight runs to completion before `stop` returns.
 
 **Reading.** `read(provider, entry_id)` returns the region's entries with id at least `entry_id`, in order, fetching only that region's segments with range reads and decoding nothing else, and hides entries at or below the region's obsolete watermark. `obsolete` records the watermark in memory, clamped to the durable entry id in the `enqueued` mode; the watermark is re-established after a restart because Mito calls `obsolete` with the manifest's `flushed_entry_id` when it opens a region, so no WAL-owned watermark state is persisted. `obsolete_all` hides the whole region. `latest_entry_id` is not affected by the watermark. Every method taking a provider requires `Provider::ObjectStore` with the store's prefix and the caller's region.
 
-**Metrics.** The store exports metrics under `greptime_logstore_object_store_wal_`. The write path reports `seal_to_durable_seconds`, the time from sealing a batch until its object is durable, `append_ack_seconds`, the time an append waits for its acknowledgement in the `durable` mode, `object_bytes` and `object_entries`, the size and the entry count of every object sealed, and `created_objects_total`, the creates that succeeded, an identical retry included; a create is counted and timed whether or not its batch is still there to be indexed, since an object a poisoned store gave up on is as durable as any other. `indexed_objects` and `indexed_bytes` are the objects and the bytes the catalog holds, set by a recovery that succeeded and moved by every object indexed or unindexed afterwards, so a collection takes the object it deleted off both, and a recovery that is abandoned publishes neither. A recovery that succeeds reports `recovery_seconds` and the objects it read footers from in `recovered_objects_total`, both of the same recoveries, so one that is abandoned at a fetch or rejected by the catalog contributes to neither; a read stream reports `read_seconds` once it has been read to its end, and one that fails or is dropped part way reports nothing. The failure paths are counted: `create_conflicts_total` for a create that found a conflicting object, `create_failures_total` for one the object store did not confirm, `poisoned_total` for a store that recorded a terminal error, `skipped_segments_total` for a segment a read skipped, and `stalled_appends_total` with `stalled_append_seconds` for an append the `enqueued` backlog thresholds held back, timed until it is admitted or refused, a stop, a terminal error or the teardown of the actor releasing it included; the delete counters of a collection are described under *Entry ids*. The histogram buckets are chosen for the default flush interval of 100 ms and the default `max_batch_bytes` of 8 MiB.
+**Metrics.** The store exports metrics under `greptime_logstore_object_store_wal_`. The write path reports `seal_to_durable_seconds`, the time from sealing a batch until its object is durable, `append_ack_seconds`, the time an append waits for its acknowledgement in the `durable` mode, `object_bytes` and `object_entries`, the size and the entry count of every object sealed, and `created_objects_total`, the creates that succeeded, an identical retry included; a create is counted and timed whether or not its batch is still there to be indexed, since an object a poisoned store gave up on is as durable as any other. `indexed_objects` and `indexed_bytes` are the objects and the bytes the catalog holds, set by a recovery that succeeded and moved by every object indexed or unindexed afterwards, so a collection takes the object it deleted off both, and a recovery that is abandoned publishes neither. A recovery that succeeds reports `recovery_seconds` and the objects it read footers from in `recovered_objects_total`, both of the same recoveries, so one that is abandoned at a fetch or rejected by the catalog contributes to neither; a read stream reports `read_seconds` once it has been read to its end, and one that fails or is dropped part way reports nothing. The failure paths are counted: `create_conflicts_total` for a create that found a conflicting object, `create_failures_total` for one the object store did not confirm, `poisoned_total` for a store that recorded a terminal error, `skipped_segments_total` for a segment a read skipped, and `stalled_appends_total` with `stalled_append_seconds` for an append held back because the sealed batches were full, in either mode, or by the `enqueued` backlog thresholds, timed until it is admitted or refused, a stop, a terminal error, a failing store or the teardown of the actor releasing it included. The gauge `failing` counts the stores whose backlog the object store refuses with an error that requires intervention, and `rejected_appends_total` the appends such a store rejected; the attempts of a failing store are counted in `create_failures_total`. The delete counters of a collection are described under *Entry ids*. The histogram buckets are chosen for the default flush interval of 100 ms and the default `max_batch_bytes` of 8 MiB.
 
 A `testing` cargo feature exposes hooks to wait for admitted appends, to seal the open batch, to hold and release creates and to make creates fail, so engine tests can flush deterministically and observe entries that are admitted but not durable.
 
@@ -233,14 +238,14 @@ The reference implementation was validated on the following evidence; the tests
 | Recovery never silently loses data | Unit coverage of every corruption class at recovery and on read; footer-only recovery compared with a full decode; a footer longer than the tail window. | Structural damage fails construction of the store; a corrupted segment is skipped, recorded as a hole of its region and counted, or fails the read under `fail`; the footer-only catalog equals the full decode. |
 | A region cannot be silently pointed at another WAL | Reopen on a mismatched prefix; reopen on an empty prefix. | The mismatch is rejected with both prefixes named and the region is not opened; the empty prefix opens cleanly with the expected row counts. |
 | Garbage collection deletes everything below the persisted reclaim boundary, and recovery stays correct while objects disappear | Unit tests of the boundary computation (a region without a watermark holds it at its first object, a watermark inside an object keeps the object, `obsolete_all` releases it, last indexed plus one when everything is flushed), of stamping and durability (headers carry the boundary current at sealing, a failed create's boundary permits nothing, the start object carries the recovered boundary), of deletion (chain objects, listed orphans and failed batches below the durable boundary, nothing at or above it, a failed delete retried, `stop` waiting for deletes in flight, holes dropped), of recovery (two restarts with a collection between them, a late object of an earlier epoch below the boundary collected, a missing predecessor above the boundary still failing) and of a read that listed an object before it was collected; the Mito flush-then-collect test in both acknowledgement modes; the standalone integration test's assertion after a flush. | Objects below the durable boundary are gone after a flush, the start object included; a restart replays everything above the boundary; exactly one WAL object remains after a flush and one more insert in the integration test. |
-| Every row of the *Failure matrix* holds, and every metric moves as documented | A table-driven fault matrix, `test_store_fault_matrix`, that drives every failure the object store access can produce through the store and asserts, per case, which waiters fail and with what error, whether the store is poisoned, whether the sequence is consumed, the indexed gauges at the failure itself, the exact movement of every metric, and what a read of the store sees afterwards; every timer is observed twice around two forced waits whose intervals cannot overlap. | 33 cases: 17 for the ten rows of the matrix and 16 for the failures listed under *Failures driven through the store*. The test fails if a row of the matrix is left uncovered. |
+| Every row of the *Failure matrix* holds, and every metric moves as documented | A table-driven fault matrix, `test_store_fault_matrix`, that drives every failure the object store access can produce through the store and asserts, per case, which waiters fail and with what error, whether the store is poisoned, whether the sequence is consumed, the indexed gauges at the failure itself, the exact movement of every metric, and what a read of the store sees afterwards; every timer is observed twice around two forced waits whose intervals cannot overlap. | 36 cases: 19 for the eleven rows of the matrix and 17 for the failures listed under *Failures driven through the store*. The test fails if a row of the matrix is left uncovered. |
 | Recovery cost does not grow with object size | Footer-only recovery over 2,000 objects; recovery concurrency bounded and ordered by sequence. | Construction of the store fell from 3.6 s to 0.9 s. |
 
 ### Failures driven through the store
 
-- Conditional create outcomes: success, identical retry, conflicting content from an earlier epoch and from the same or a later one, lost response, transient failure in `durable` mode with later objects already created and with creates still in flight (every batch that is not indexed fails, no sequence is reused, late completions are orphans), transient failure in `enqueued` mode and after `stop` began, completion after a conflict had already poisoned the store, a create at the last representable sequence.
+- Conditional create outcomes: success, identical retry, conflicting content from an earlier epoch and from the same or a later one, lost response, transient failure in `durable` mode with later objects already created and with creates still in flight (every batch that is not indexed fails, no sequence is reused, late completions are orphans), transient failure in `enqueued` mode and after `stop` began, a failure that requires intervention in `enqueued` mode and a `stop` while it lasts, completion after a conflict had already poisoned the store, a create at the last representable sequence.
 - Object store access: an unwritable store; a listing, a trailer fetch, a footer fetch, a segment fetch and a delete that fail; a delete that fails and is retried by the next collection; an object collected between the listing and its fetch, below and at the tip's reclaim boundary; an object store that answers slowly.
-- Store state: an object the catalog rejects, per-region entry id exhaustion, the catalog refusing a next sequence past the maximum, every stop path with several creates in flight that succeed, fail transiently or conflict after stop began, a stop with a delete in flight, and the appends the backlog thresholds were holding back released by admission, by a stop, by a terminal error and by the teardown of the actor.
+- Store state: an object the catalog rejects, per-region entry id exhaustion, the catalog refusing a next sequence past the maximum, every stop path with several creates in flight that succeed, fail transiently or conflict after stop began, a stop with a delete in flight, an append held back by full sealed batches, and the appends the backlog thresholds were holding back released by admission, by a stop, by a terminal error and by the teardown of the actor.
 - Pipelined uploader: concurrent creates acknowledged in sequence order, every sealed batch linked to its pipeline predecessor, the open batch sealing at the threshold while creates are in flight, a transient failure failing every batch that is not indexed, created ones included, a permanent failure poisoning the store.
 - Chain recovery: a linear chain, orphans below and above the tip, a broken tip falling back to the latest complete object, a late object of an earlier epoch next to the chain of a later one, a missing predecessor, objects present without a complete chain, orphans raising the next sequence, and the start object on an empty prefix and meeting objects of an earlier and of a later epoch.
 

src/log-store/src/error.rs

@@ -436,6 +436,15 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display(
+        "Object store WAL rejects appends until the object store accepts its acknowledged backlog"
+    ))]
+    ObjectStoreWalFailing {
+        source: Arc<Error>,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -495,6 +504,7 @@ impl ErrorExt for Error {
 
             InvalidWalObject { source, .. } => source.status_code(),
             ObjectStoreWal { source, .. } => source.status_code(),
+            ObjectStoreWalFailing { .. } => StatusCode::StorageUnavailable,
 
             // Object store related errors
             WalObjectStore { .. }
@@ -534,7 +544,8 @@ impl ErrorExt for Error {
             | AddEntryLogBatch { .. }
             | WalObjectSequenceUnsettled { .. }
             | StaleWalObject { .. }
-            | UnconfirmedWalEpochStart { .. } => RetryHint::Retryable,
+            | UnconfirmedWalEpochStart { .. }
+            | ObjectStoreWalFailing { .. } => RetryHint::Retryable,
             ProduceRecord { error, .. } => match error {
                 rskafka::client::producer::Error::Client(error) => {
                     rskafka_client_error_to_retry_hint(error)

src/log-store/src/metrics.rs

@@ -244,7 +244,7 @@ lazy_static! {
     /// `durable` mode and are repeated in the `enqueued` mode.
     pub static ref METRIC_OBJECT_STORE_WAL_CREATE_FAILURES_TOTAL: IntCounter = register_int_counter!(
         "greptime_logstore_object_store_wal_create_failures_total",
-        "object store logstore transient conditional create failures total",
+        "object store logstore conditional creates the object store did not confirm total",
     )
     .unwrap();
 
@@ -256,12 +256,14 @@ lazy_static! {
     )
     .unwrap();
 
-    /// Counter of appends the object store logstore held back because the
-    /// unpersisted backlog reached a threshold in the `enqueued` mode, which
-    /// says the write rate is above what the uploads sustain.
+    /// Counter of appends the object store logstore held back until an upload
+    /// completed: in either mode because the sealed batches waiting to become
+    /// durable left no room for another, and in the `enqueued` mode also
+    /// because the unpersisted backlog reached a threshold. Either says the
+    /// write rate is above what the uploads sustain.
     pub static ref METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL: IntCounter = register_int_counter!(
         "greptime_logstore_object_store_wal_stalled_appends_total",
-        "object store logstore appends stalled by the backlog thresholds total",
+        "object store logstore appends stalled by full sealed batches or the backlog thresholds total",
     )
     .unwrap();
 
@@ -278,4 +280,24 @@ lazy_static! {
         exponential_buckets(0.005, 3.0, 9).unwrap(),
     )
     .unwrap();
+
+    /// Number of object store logstores in the `enqueued` mode whose backlog
+    /// the object store refuses with an error that repeating the create
+    /// cannot settle, such as denied access or a missing bucket. Such a store
+    /// keeps its backlog, attempts it again with backoff and rejects appends
+    /// until a create succeeds.
+    pub static ref METRIC_OBJECT_STORE_WAL_FAILING: IntGauge = register_int_gauge!(
+        "greptime_logstore_object_store_wal_failing",
+        "object store logstores rejecting appends because the object store refuses their backlog with an error that requires intervention",
+    )
+    .unwrap();
+
+    /// Counter of appends the object store logstore rejected while its
+    /// backlog is refused, see the gauge above. They take no entry id and
+    /// fail with a retryable error.
+    pub static ref METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL: IntCounter = register_int_counter!(
+        "greptime_logstore_object_store_wal_rejected_appends_total",
+        "object store logstore appends rejected while the object store refuses the backlog total",
+    )
+    .unwrap();
 }

src/mito2/src/engine/object_store_wal_recovery_test.rs

@@ -22,12 +22,12 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::sync::atomic::Ordering;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use api::v1::Rows;
 use common_base::readable_size::ReadableSize;
-use common_error::ext::{BoxedError, ErrorExt};
+use common_error::ext::{BoxedError, ErrorExt, RetryHint};
 use common_error::status_code::StatusCode;
 use common_recordbatch::RecordBatches;
 use common_wal::config::object_store::{
@@ -35,9 +35,12 @@ use common_wal::config::object_store::{
 };
 use common_wal::options::{ObjectStoreWalOptions, WAL_OPTIONS_KEY, WalOptions};
 use log_store::ObjectStoreLogStore;
+use log_store::metrics::METRIC_OBJECT_STORE_WAL_FAILING;
 use log_store::object_store_wal::{WalHole, entry_id};
 use object_store::ObjectStore;
+use object_store::layers::mock::{self, MockLayerBuilder, oio};
 use object_store::services::Memory;
+use object_store::util::with_retry_layers;
 use rstest::rstest;
 use store_api::logstore::LogStore;
 use store_api::logstore::provider::Provider;
@@ -1355,3 +1358,106 @@ async fn test_flush_does_not_publish_a_frontier_for_a_lost_enqueued_backlog() {
     );
     assert!(store.stop().await.is_err());
 }
+
+/// An in-memory object store behind the retry layer object storage has in
+/// production, whose writes fail with denied access while `denied` is set.
+fn denying_store(denied: Arc<AtomicBool>) -> ObjectStore {
+    let mock = MockLayerBuilder::default()
+        .writer_factory(Arc::new(move |_, _, inner| {
+            if denied.load(Ordering::SeqCst) {
+                Box::new(DeniedWriter)
+            } else {
+                inner
+            }
+        }))
+        .build()
+        .unwrap();
+    with_retry_layers(memory_store().layer(mock))
+}
+
+struct DeniedWriter;
+
+impl oio::Write for DeniedWriter {
+    async fn write(&mut self, _buffer: mock::Buffer) -> mock::Result<()> {
+        Ok(())
+    }
+
+    async fn close(&mut self) -> mock::Result<mock::Metadata> {
+        Err(mock::Error::new(
+            mock::ErrorKind::PermissionDenied,
+            "access denied",
+        ))
+    }
+
+    async fn abort(&mut self) -> mock::Result<()> {
+        Ok(())
+    }
+}
+
+#[tokio::test]
+async fn test_enqueued_writes_fail_fast_while_the_object_store_refuses_the_backlog() {
+    let mut env = TestEnv::with_prefix("object-store-wal-refused-backlog").await;
+    let denied = Arc::new(AtomicBool::new(false));
+    let store = open_store(&denying_store(denied.clone()), AckMode::Enqueued).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    let (_, schema) = create_region(&engine, REGION_A, &[]).await;
+
+    // The write is acknowledged, then the object store denies the create of
+    // its object, which the store keeps.
+    denied.store(true, Ordering::SeqCst);
+    put_rows(&engine, REGION_A, rows(&schema, 0, 3)).await;
+    let seal = spawn_seal(&store);
+    tokio::time::timeout(WAIT, async {
+        while METRIC_OBJECT_STORE_WAL_FAILING.get() == 0 {
+            tokio::time::sleep(Duration::from_millis(1)).await;
+        }
+    })
+    .await
+    .expect("the store must fail");
+
+    // A later write fails at once with a retryable error that names the
+    // refusal, and takes no entry id.
+    let error = try_put(&engine, REGION_A, rows(&schema, 3, 5))
+        .await
+        .unwrap_err();
+    assert_eq!(StatusCode::StorageUnavailable, error.status_code());
+    assert_eq!(RetryHint::Retryable, error.retry_hint());
+    assert!(
+        format!("{error:?}").contains("PermissionDenied"),
+        "{error:?}"
+    );
+    assert_eq!(
+        entry_id(1, 1),
+        region_recovery_state(&engine, REGION_A).await.last_entry_id
+    );
+
+    // A flush waits for the acknowledged entry instead of failing.
+    let flush = spawn_flush(&engine, REGION_A);
+    wait_for(store.wait_for_durability_waits(1)).await;
+    assert!(!flush.is_finished());
+
+    // Once the object store accepts writes again, the backlog is uploaded,
+    // the flush records the entry and writes are accepted again.
+    denied.store(false, Ordering::SeqCst);
+    tokio::time::timeout(WAIT, seal)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    tokio::time::timeout(WAIT, flush)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    put_rows(&engine, REGION_A, rows(&schema, 3, 5)).await;
+    assert_eq!(
+        RegionRecoveryState {
+            flushed_entry_id: entry_id(1, 1),
+            last_entry_id: entry_id(2, 1),
+            topic_latest_entry_id: entry_id(1, 1),
+            manifest_flushed_entry_id: entry_id(1, 1),
+            memtable_rows: 2,
+        },
+        region_recovery_state(&engine, REGION_A).await
+    );
+}