diff --git a/docs/rfcs/2026-09-06-object-store-wal.md b/docs/rfcs/2026-09-06-object-store-wal.md index 26970d166f8..b49f97c330c 100644 --- a/docs/rfcs/2026-09-06-object-store-wal.md +++ b/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. diff --git a/src/log-store/src/error.rs b/src/log-store/src/error.rs index 836d1fbfe7d..c7b9c981c44 100644 --- a/src/log-store/src/error.rs +++ b/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, + #[snafu(implicit)] + location: Location, + }, } pub type Result = std::result::Result; @@ -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) diff --git a/src/log-store/src/metrics.rs b/src/log-store/src/metrics.rs index 03a3dc1f174..c787820aaeb 100644 --- a/src/log-store/src/metrics.rs +++ b/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(); } diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index 530f76b9ff6..f233f73bdea 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -24,12 +24,12 @@ use std::time::{Duration, Instant}; use async_stream::try_stream; use bytes::Bytes; -use common_telemetry::{debug, info, warn}; +use common_telemetry::{debug, error, info, warn}; use common_wal::config::object_store::{AckMode, CorruptedSegmentAction, ObjectStoreWalConfig}; -use futures::future::BoxFuture; +use futures::future::{BoxFuture, OptionFuture}; use futures::stream::FuturesUnordered; use futures::{StreamExt, TryStreamExt}; -use object_store::ObjectStore; +use object_store::{ErrorKind, ObjectStore}; use snafu::{IntoError, OptionExt, ResultExt, ensure}; use store_api::logstore::entry::{Entry, NaiveEntry}; use store_api::logstore::provider::{ObjectStoreProvider, Provider}; @@ -42,18 +42,19 @@ use tokio_util::task::AbortOnDropHandle; use crate::error::{ CorruptedWalObjectSnafu, Error, IncompleteWalEntrySnafu, InvalidProviderSnafu, InvalidWalObjectSnafu, InvalidWalObjectStoreSnafu, MismatchedWalPrefixSnafu, - MismatchedWalRegionSnafu, ObjectStoreWalSnafu, ObjectStoreWalStoppedSnafu, Result, - StaleWalObjectSnafu, UnconfirmedWalEpochStartSnafu, WalObjectSequenceExhaustedSnafu, - WalObjectSequenceUnsettledSnafu, WalSegmentFetchSnafu, + MismatchedWalRegionSnafu, ObjectStoreWalFailingSnafu, ObjectStoreWalSnafu, + ObjectStoreWalStoppedSnafu, Result, StaleWalObjectSnafu, UnconfirmedWalEpochStartSnafu, + WalObjectSequenceExhaustedSnafu, WalObjectSequenceUnsettledSnafu, WalSegmentFetchSnafu, }; use crate::metrics::{ METRIC_OBJECT_STORE_WAL_APPEND_ACK_SECONDS, METRIC_OBJECT_STORE_WAL_CREATE_CONFLICTS_TOTAL, METRIC_OBJECT_STORE_WAL_CREATE_FAILURES_TOTAL, METRIC_OBJECT_STORE_WAL_CREATED_OBJECTS_TOTAL, METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL, METRIC_OBJECT_STORE_WAL_FAILED_DELETES_TOTAL, - METRIC_OBJECT_STORE_WAL_INDEXED_BYTES, METRIC_OBJECT_STORE_WAL_INDEXED_OBJECTS, - METRIC_OBJECT_STORE_WAL_OBJECT_BYTES, METRIC_OBJECT_STORE_WAL_OBJECT_ENTRIES, - METRIC_OBJECT_STORE_WAL_POISONED_TOTAL, METRIC_OBJECT_STORE_WAL_READ_SECONDS, - METRIC_OBJECT_STORE_WAL_RECOVERED_OBJECTS_TOTAL, METRIC_OBJECT_STORE_WAL_RECOVERY_SECONDS, + METRIC_OBJECT_STORE_WAL_FAILING, METRIC_OBJECT_STORE_WAL_INDEXED_BYTES, + METRIC_OBJECT_STORE_WAL_INDEXED_OBJECTS, METRIC_OBJECT_STORE_WAL_OBJECT_BYTES, + METRIC_OBJECT_STORE_WAL_OBJECT_ENTRIES, METRIC_OBJECT_STORE_WAL_POISONED_TOTAL, + METRIC_OBJECT_STORE_WAL_READ_SECONDS, METRIC_OBJECT_STORE_WAL_RECOVERED_OBJECTS_TOTAL, + METRIC_OBJECT_STORE_WAL_RECOVERY_SECONDS, METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL, METRIC_OBJECT_STORE_WAL_SEAL_TO_DURABLE_SECONDS, METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL, METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS, METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL, @@ -76,11 +77,21 @@ const MAX_IN_FLIGHT_DELETES: usize = 4; /// Delay before a create that failed transiently is attempted again in the /// `enqueued` acknowledgement mode, where no caller is left to retry it. const CREATE_RETRY_DELAY: Duration = Duration::from_millis(100); +/// Longest delay from a failed attempt of a backlog that the object store +/// refuses with an error that requires intervention, see [`Failing`], to the +/// next attempt. +/// The delay doubles from [`CREATE_RETRY_DELAY`] after every attempt. At the +/// cap a dead bucket costs four requests a minute, a conditional create and +/// the read that reconciles it every 30 seconds; a longer cap would keep +/// appends rejected longer after the cause is fixed. +const MAX_FAILING_RETRY_DELAY: Duration = Duration::from_secs(30); +/// How often a store logs that the object store still refuses its backlog. +const FAILING_LOG_INTERVAL: Duration = Duration::from_secs(60); /// Number of sealed batches that may wait to be created or indexed. One /// admission can seal two batches, the open batch before an append that would -/// exhaust its positions and the append's own batch, so the actor takes an -/// append from its channel only while two more fit: a stalled object store -/// holds callers back instead of growing the backlog. +/// exhaust its positions and the append's own batch, so the actor admits an +/// append only while two more fit and holds it back otherwise: a stalled +/// object store holds callers back instead of growing the backlog. const MAX_SEALED_BATCHES: usize = 2 * MAX_IN_FLIGHT_CREATES; /// Number of objects whose footers recovery fetches at a time. On S3 an /// object costs two reads, its header and its tail, tens of milliseconds per @@ -123,7 +134,9 @@ const RECOVERY_TAIL_WINDOW: usize = 64 * 1024; /// indexed, so an acknowledged entry never has a missing predecessor; in the /// `enqueued` mode it returns on admission and the object is created in the /// background. While [`MAX_SEALED_BATCHES`] batches -/// wait to become durable no append is admitted. A segment that a read cannot +/// wait to become durable no append is admitted, and in the `enqueued` mode +/// appends are rejected while the object store refuses the backlog with an +/// error that requires intervention. A segment that a read cannot /// decode is handled as `on_corrupted_segment` says. Objects below the reclaim /// boundary of the last indexed object are deleted in the background. pub struct ObjectStoreLogStore { @@ -377,6 +390,7 @@ impl ObjectStoreLogStore { sealed: VecDeque::new(), creates: FuturesUnordered::new(), stalled: None, + failing: None, durable_waiters: Vec::new(), stop: Vec::new(), stop_error: None, @@ -1046,7 +1060,8 @@ struct DurableWaiter { /// Where the conditional create of a sealed batch stands. enum CreateState { /// The create has not started: no slot was free, or in the `enqueued` - /// mode an earlier attempt failed transiently and is repeated. + /// mode an earlier attempt failed with an object store error and is + /// repeated. Pending, InFlight, Created, @@ -1087,6 +1102,48 @@ impl SealedBatch { type CreateOutcome = (u64, Result); +/// The state of an `enqueued` store whose backlog the object store refuses +/// with an error that requires intervention, see [`requires_intervention`], +/// such as denied +/// access or a missing bucket. +/// +/// Poisoning would drop the acknowledged backlog and need a restart, while +/// such a cause can be fixed in place. So the backlog keeps its sequences and +/// bytes, and its lowest pending batch is attempted alone, once nothing is in +/// flight, after a delay that doubles from [`CREATE_RETRY_DELAY`] up to +/// [`MAX_FAILING_RETRY_DELAY`]. Appends are rejected at once with a retryable +/// error that carries the last error: they take no entry id and never wait +/// behind the backlog. The first create that succeeds ends the state. +struct Failing { + /// The last error a create met, which rejected appends carry. + error: Arc, + since: Instant, + /// The delay before the next attempt is due once nothing is in flight. + delay: Duration, + /// When the next attempt starts, `None` until it is scheduled. + probe_at: Option, + failed_creates: u64, + rejected_appends: u64, + logged_at: Instant, +} + +impl Failing { + /// Schedules the next attempt after the current delay and doubles the + /// delay for the one after it. + fn schedule(&mut self) { + self.probe_at = Some(Instant::now() + self.delay); + self.delay = (self.delay * 2).min(MAX_FAILING_RETRY_DELAY); + } + + /// Counts an admission the store refuses and returns the error it + /// fails with. + fn reject(&mut self) -> Error { + self.rejected_appends += 1; + METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.inc(); + ObjectStoreWalFailingSnafu.into_error(self.error.clone()) + } +} + /// The actor that owns the open batch and the sealed batches until they are /// durable. /// @@ -1103,7 +1160,8 @@ type CreateOutcome = (u64, Result); /// | created, or identical retry | advances | acknowledged in order | healthy | /// | transient error, `durable` mode | never reused: a create that reported an error may have written its object | every batch that is not indexed fails, created ones included; retries are assigned new ids | healthy: the next batch links to the last indexed object, so no chain reaches a failed batch | /// | transient error, `enqueued` mode | unchanged | already acknowledged | healthy: the create is repeated with the same bytes after [`CREATE_RETRY_DELAY`], which an identical retry accepts | -/// | transient error, `enqueued` mode after `stop` began | never reused | already acknowledged | the backlog is dropped and `stop` reports the error | +/// | error that requires intervention, `enqueued` mode | unchanged | already acknowledged; later appends are rejected with a retryable error | failing, see [`Failing`]: the lowest pending create is repeated alone with backoff, and the first create that succeeds ends it | +/// | object store error, `enqueued` mode after `stop` began | never reused | already acknowledged | the backlog is dropped and `stop` reports the error | /// | an object of an earlier epoch holds the sequence, `durable` mode | never reused | as for a transient error | healthy: that object can never be on the chain | /// | an object of the same or a later epoch holds the sequence, any conflicting object in the `enqueued` mode, encoding or catalog error | unchanged | every batch that is not indexed fails | poisoned | /// | created at the last representable sequence | cannot advance | acknowledged | poisoned: no later batch can be allocated a sequence | @@ -1114,10 +1172,11 @@ type CreateOutcome = (u64, Result); /// uploads its backlog; creates in flight run to completion and acknowledge /// if they succeed. /// -/// Appends arrive on their own channel, which the actor does not read while -/// [`MAX_SEALED_BATCHES`] batches wait or an append is held back at a backlog -/// threshold of the `enqueued` mode, so commands such as `stop` are handled -/// while admission is held back. +/// Appends arrive on their own channel. The actor holds one back while the +/// sealed batches leave no room for the two an admission can seal, or at a +/// backlog threshold of the `enqueued` mode, and does not read the channel +/// meanwhile, so commands such as `stop` are handled while admission is held +/// back. A failing store reads the channel to reject every append. /// /// The actor also collects garbage. Every batch carries the reclaim boundary /// current when it was sealed, see [`ObjectCatalog::reclaim_boundary`], which @@ -1152,10 +1211,13 @@ struct Actor { sealed: VecDeque, /// Creates in flight, including those of batches that already failed. creates: FuturesUnordered>, - /// The append held back in the `enqueued` mode while the unpersisted - /// backlog is at a threshold, and when it was held back. Later appends - /// wait in their channel. + /// The append held back while the sealed batches are full or, in the + /// `enqueued` mode, the unpersisted backlog is at a threshold, and when + /// it was held back. Later appends wait in their channel. stalled: Option<(QueuedAppend, Instant)>, + /// Set while the object store refuses the `enqueued` backlog with an + /// error that requires intervention. + failing: Option, durable_waiters: Vec, /// Callers of `stop`, answered once nothing is in flight. stop: Vec>>, @@ -1207,6 +1269,7 @@ impl Actor { interval.tick().await; loop { + let probe_at = self.failing.as_ref().and_then(|failing| failing.probe_at); tokio::select! { _ = interval.tick() => { // Nothing starts after stop began; the `enqueued` mode @@ -1221,7 +1284,10 @@ impl Actor { Some((object_seq, result)) = self.deletes.next(), if !self.deletes.is_empty() => { self.on_delete_completed(object_seq, result); } - Some((entries, response)) = self.append_rx.recv(), if self.stalled.is_none() && self.sealed.len() + 2 <= MAX_SEALED_BATCHES => { + Some(()) = OptionFuture::from(probe_at.map(|at| tokio::time::sleep_until(at.into()))) => { + self.probe(); + } + Some((entries, response)) = self.append_rx.recv(), if self.stalled.is_none() => { self.handle_append(entries, response); } command = self.command_rx.recv() => match command { @@ -1271,7 +1337,11 @@ impl Actor { let _ = response.send(Err(shared(&error))); return; } - if self.ack_mode == AckMode::Enqueued && self.backlog_at_threshold() { + if let Some(failing) = &mut self.failing { + let _ = response.send(Err(failing.reject())); + return; + } + if self.should_hold_back() { METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.inc(); self.stalled = Some(((entries, response), Instant::now())); self.ensure_create_in_flight(); @@ -1336,6 +1406,14 @@ impl Actor { } } + /// Returns true while an append has to wait for an upload: the sealed + /// batches leave no room for the two an admission can seal, or in the + /// `enqueued` mode the unpersisted backlog is at a threshold. + fn should_hold_back(&self) -> bool { + self.sealed.len() + 2 > MAX_SEALED_BATCHES + || (self.ack_mode == AckMode::Enqueued && self.backlog_at_threshold()) + } + /// Returns true once the unpersisted backlog, the open batch and every /// sealed batch that is not durable, reaches the size or the age threshold. fn backlog_at_threshold(&self) -> bool { @@ -1364,8 +1442,7 @@ impl Actor { } } - /// Admits the stalled append once the backlog is below the thresholds and - /// the sealed batches leave room for the two an admission can seal. + /// Admits the stalled append once it no longer has to be held back. fn release_stalled(&mut self) { if self.stalled.is_none() { return; @@ -1376,14 +1453,11 @@ impl Actor { } return; } - if self.backlog_at_threshold() { + if self.should_hold_back() { // The append needs an upload to complete. self.ensure_create_in_flight(); return; } - if self.sealed.len() + 2 > MAX_SEALED_BATCHES { - return; - } if let Some((entries, response)) = self.take_stalled() { self.admit(entries, response); } @@ -1475,88 +1549,122 @@ impl Actor { /// Starts the creates of pending batches in sequence order while fewer /// than [`MAX_IN_FLIGHT_CREATES`] are in flight, counting the creates of /// batches that already failed. Nothing starts once stop began, except - /// the backlog of the `enqueued` mode. + /// the backlog of the `enqueued` mode. A failing store only schedules the + /// next attempt once nothing is in flight, see [`Failing`]; after stop + /// began it uploads its backlog at once. fn start_creates(&mut self) { if self.is_stopped() && self.ack_mode == AckMode::Durable { return; } + if !self.is_stopped() + && let Some(failing) = &mut self.failing + { + if self.creates.is_empty() && failing.probe_at.is_none() { + failing.schedule(); + } + return; + } let mut in_flight = self.creates.len(); - for batch in self.sealed.iter_mut() { + for index in 0..self.sealed.len() { if in_flight >= MAX_IN_FLIGHT_CREATES { break; } + let batch = &self.sealed[index]; if !matches!(batch.state, CreateState::Pending) { continue; } - batch.state = CreateState::InFlight; - in_flight += 1; let delay = if batch.attempts == 0 { Duration::ZERO } else { CREATE_RETRY_DELAY }; - batch.attempts += 1; - let io = self.io.clone(); - let object_seq = batch.object_seq; - let sealed_at = batch.sealed_at; - self.creating.insert(object_seq); - let bytes = batch.bytes.clone(); - let epoch = self.epoch; - let ack_mode = self.ack_mode; - #[cfg(any(test, feature = "testing"))] - let mut creates_held = self.creates_held.clone(); - #[cfg(any(test, feature = "testing"))] - let creates_fail = self.creates_fail.clone(); - #[cfg(any(test, feature = "testing"))] - let next_create_fails_after_write = self.next_create_fails_after_write.clone(); - #[cfg(any(test, feature = "testing"))] - let parked_creates = self.parked_creates.clone(); - self.creates.push(Box::pin(async move { - if !delay.is_zero() { - tokio::time::sleep(delay).await; - } - // The store was dropped while the create was parked: it never - // runs. - #[cfg(any(test, feature = "testing"))] - if *creates_held.borrow() { - parked_creates.send_modify(|count| *count += 1); - } - #[cfg(any(test, feature = "testing"))] - if creates_held.wait_for(|held| !*held).await.is_err() { - return (object_seq, Err(ObjectStoreWalStoppedSnafu.build())); - } - #[cfg(any(test, feature = "testing"))] - if creates_fail.load(Ordering::Acquire) { - let result = injected_create_failure(io.as_ref(), object_seq); - record_create(&result, Some(sealed_at)); - return (object_seq, result); - } - let created = io.put_if_absent(object_seq, bytes).await; - #[cfg(any(test, feature = "testing"))] - let created = match created { - Ok(_) if next_create_fails_after_write.swap(false, Ordering::AcqRel) => { - injected_create_failure(io.as_ref(), object_seq) - } - created => created, - }; - // Counted as the object store answered, before the epoch of a - // conflicting object is read. - record_create(&created, Some(sealed_at)); - // Any conflict poisons an `enqueued` store, so only the - // `durable` mode reads the epoch of the existing object. - let result = match created { - Err(error @ Error::WalObjectConflict { .. }) - if ack_mode == AckMode::Durable => - { - stale_conflict(io.as_ref(), object_seq, epoch, error).await - } - result => result, - }; - (object_seq, result) - })); + self.start_create(index, delay); + in_flight += 1; } } + /// Attempts the lowest pending batch of a failing store, whose delay + /// elapsed. After stop began `start_creates` uploads the backlog instead. + fn probe(&mut self) { + let Some(failing) = &mut self.failing else { + return; + }; + failing.probe_at = None; + if self.is_stopped() { + return; + } + if let Some(index) = self + .sealed + .iter() + .position(|batch| matches!(batch.state, CreateState::Pending)) + { + self.start_create(index, Duration::ZERO); + } + } + + /// Starts the create of the sealed batch at `index` after `delay`. + fn start_create(&mut self, index: usize, delay: Duration) { + let batch = &mut self.sealed[index]; + batch.state = CreateState::InFlight; + batch.attempts += 1; + let object_seq = batch.object_seq; + let sealed_at = batch.sealed_at; + let bytes = batch.bytes.clone(); + let io = self.io.clone(); + self.creating.insert(object_seq); + let epoch = self.epoch; + let ack_mode = self.ack_mode; + #[cfg(any(test, feature = "testing"))] + let mut creates_held = self.creates_held.clone(); + #[cfg(any(test, feature = "testing"))] + let creates_fail = self.creates_fail.clone(); + #[cfg(any(test, feature = "testing"))] + let next_create_fails_after_write = self.next_create_fails_after_write.clone(); + #[cfg(any(test, feature = "testing"))] + let parked_creates = self.parked_creates.clone(); + self.creates.push(Box::pin(async move { + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + // The store was dropped while the create was parked: it never + // runs. + #[cfg(any(test, feature = "testing"))] + if *creates_held.borrow() { + parked_creates.send_modify(|count| *count += 1); + } + #[cfg(any(test, feature = "testing"))] + if creates_held.wait_for(|held| !*held).await.is_err() { + return (object_seq, Err(ObjectStoreWalStoppedSnafu.build())); + } + #[cfg(any(test, feature = "testing"))] + if creates_fail.load(Ordering::Acquire) { + let result = injected_create_failure(io.as_ref(), object_seq); + record_create(&result, Some(sealed_at)); + return (object_seq, result); + } + let created = io.put_if_absent(object_seq, bytes).await; + #[cfg(any(test, feature = "testing"))] + let created = match created { + Ok(_) if next_create_fails_after_write.swap(false, Ordering::AcqRel) => { + injected_create_failure(io.as_ref(), object_seq) + } + created => created, + }; + // Counted as the object store answered, before the epoch of a + // conflicting object is read. + record_create(&created, Some(sealed_at)); + // Any conflict poisons an `enqueued` store, so only the + // `durable` mode reads the epoch of the existing object. + let result = match created { + Err(error @ Error::WalObjectConflict { .. }) if ack_mode == AckMode::Durable => { + stale_conflict(io.as_ref(), object_seq, epoch, error).await + } + result => result, + }; + (object_seq, result) + })); + } + fn on_create_completed(&mut self, object_seq: u64, result: Result) { self.creating.remove(&object_seq); // A batch that already failed, or that the store gave up on when it @@ -1572,28 +1680,27 @@ impl Actor { return; }; match result { - Ok(_) => self.sealed[index].state = CreateState::Created, + Ok(_) => { + self.sealed[index].state = CreateState::Created; + self.leave_failing("a create succeeded"); + } // Nobody is left to retry in the `enqueued` mode, so the store - // repeats a create that failed transiently under the same sequence - // with the same bytes, which an identical retry accepts. - Err(ref error) - if self.ack_mode == AckMode::Enqueued - && !self.is_stopped() - && is_transient(error) => + // repeats a create that the object store did not confirm under + // the same sequence with the same bytes, which an identical retry + // accepts: after `CREATE_RETRY_DELAY`, or with backoff while the + // store is failing. + Err(error @ Error::WalObjectStore { .. }) + if self.ack_mode == AckMode::Enqueued && !self.is_stopped() => { self.sealed[index].state = CreateState::Pending; + self.on_create_failed(error); } // The object store did not confirm the object, which may still // exist or land later, or an earlier epoch holds the sequence and // can never be on the chain. A caller of the `durable` mode - // retries the append itself; after stop began a transient failure - // drops the `enqueued` backlog. - Err(error @ Error::WalObjectStore { .. }) - if self.ack_mode == AckMode::Durable - || (self.is_stopped() && is_transient(&error)) => - { - self.roll_back(Arc::new(error)) - } + // retries the append itself; after stop began a failure drops the + // `enqueued` backlog. + Err(error @ Error::WalObjectStore { .. }) => self.roll_back(Arc::new(error)), Err(error @ Error::StaleWalObject { .. }) if self.ack_mode == AckMode::Durable => { self.roll_back(Arc::new(error)) } @@ -1605,6 +1712,70 @@ impl Actor { self.settle(); } + /// Records the failure of a create the `enqueued` mode repeats. An error + /// that requires intervention makes the store failing; while it is, every + /// error becomes the one rejected appends carry. + fn on_create_failed(&mut self, error: Error) { + let Some(failing) = &mut self.failing else { + if requires_intervention(&error) { + self.enter_failing(error); + } + return; + }; + failing.failed_creates += 1; + failing.error = Arc::new(error); + if failing.logged_at.elapsed() >= FAILING_LOG_INTERVAL { + failing.logged_at = Instant::now(); + warn!( + failing.error; + "Object store WAL still cannot upload its backlog of {} objects after {:?}: {} creates failed, {} appends rejected, next attempt in {:?}", + self.sealed.len(), + failing.since.elapsed(), + failing.failed_creates, + failing.rejected_appends, + failing.delay + ); + } + } + + fn enter_failing(&mut self, error: Error) { + error!( + error; + "Object store WAL cannot upload its backlog of {} objects: the object store refused a create with an error that repeating it cannot settle. Appends are rejected until a create succeeds; the backlog is kept and attempted again with backoff", + self.sealed.len() + ); + METRIC_OBJECT_STORE_WAL_FAILING.inc(); + // The append held back would wait for an upload that is not coming. + let stalled = self.take_stalled(); + let now = Instant::now(); + let failing = self.failing.insert(Failing { + error: Arc::new(error), + since: now, + delay: CREATE_RETRY_DELAY, + probe_at: None, + failed_creates: 1, + rejected_appends: 0, + logged_at: now, + }); + if let Some((_, response)) = stalled { + let _ = response.send(Err(failing.reject())); + } + } + + /// Ends the failing state, if the store is in it, because of `reason`. + fn leave_failing(&mut self, reason: &str) { + let Some(failing) = self.failing.take() else { + return; + }; + METRIC_OBJECT_STORE_WAL_FAILING.dec(); + info!( + "Object store WAL is no longer failing, {reason}, after {:?}: {} creates failed, {} appends rejected", + failing.since.elapsed(), + failing.failed_creates, + failing.rejected_appends + ); + } + /// Indexes and acknowledges the sealed batches from the front as far as /// they are created, and starts the creates that a free slot allows. fn settle(&mut self) { @@ -1718,12 +1889,15 @@ impl Actor { batch.fail(failure); } self.reset_open_batch(); - self.fail_unacknowledged(failure); + // The stalled append was never admitted; the sealed batches are gone, + // so `settle` admits it, or refuses it once stop began. + self.fail_waiters(failure); // An acknowledged backlog was dropped: `stop` reports it, whether // its caller has arrived yet or not. if self.ack_mode == AckMode::Enqueued { self.stop_error.get_or_insert(error); } + self.leave_failing("its backlog was dropped"); } /// Records `error` as terminal and fails every waiter that is not @@ -1751,6 +1925,7 @@ impl Actor { if self.ack_mode == AckMode::Enqueued { self.stop_error.get_or_insert(error.clone()); } + self.leave_failing("the store is poisoned"); error } @@ -1763,12 +1938,17 @@ impl Actor { /// Fails the waiters of the open batch, the stalled append and the /// durability waiters. fn fail_unacknowledged(&mut self, error: impl Fn() -> Error) { - for pending in self.pending.drain(..) { - let _ = pending.response.send(Err(error())); - } if let Some((_, response)) = self.take_stalled() { let _ = response.send(Err(error())); } + self.fail_waiters(error); + } + + /// Fails the waiters of the open batch and the durability waiters. + fn fail_waiters(&mut self, error: impl Fn() -> Error) { + for pending in self.pending.drain(..) { + let _ = pending.response.send(Err(error())); + } for waiter in self.durable_waiters.drain(..) { let _ = waiter.response.send(Err(error())); } @@ -1943,10 +2123,10 @@ impl Actor { return Err(shared(&error)); } // In the `enqueued` mode an id the store handed out is never handed - // out again while it runs: a create that fails transiently is - // repeated under its sequence and a permanent failure poisons the - // store. Every later id of the region is greater, so the id needs no - // floor even before it is durable. + // out again while it runs: a create that the object store did not + // confirm is repeated under its sequence and any other failure + // poisons the store. Every later id of the region is greater, so the + // id needs no floor even before it is durable. if self.ack_mode == AckMode::Enqueued && entry_id <= self.issued_entry_ids.get(®ion_id).copied().unwrap_or(0) { @@ -1980,12 +2160,16 @@ impl Actor { } } - /// Begins stopping. Nothing is admitted from now on; the `durable` mode - /// drops the open batch and the batches whose create has not started, - /// the `enqueued` mode seals its backlog so it is uploaded. Stop is - /// answered by [`finish_stop`](Self::finish_stop) once nothing is in flight. + /// Begins stopping. Nothing is admitted from now on and the append held + /// back is refused; the `durable` mode drops the open batch and the + /// batches whose create has not started, the `enqueued` mode seals its + /// backlog and uploads it, a failing store included. Stop is answered by + /// [`finish_stop`](Self::finish_stop) once nothing is in flight. fn handle_stop(&mut self, response: oneshot::Sender>) { self.stop.push(response); + if let Some((_, response)) = self.take_stalled() { + let _ = response.send(Err(ObjectStoreWalStoppedSnafu.build())); + } match self.ack_mode { AckMode::Durable => { self.reset_open_batch(); @@ -2005,10 +2189,8 @@ impl Actor { } } AckMode::Enqueued => { - if let Some((_, response)) = self.take_stalled() { - let _ = response.send(Err(ObjectStoreWalStoppedSnafu.build())); - } self.flush_open_batch(); + self.start_creates(); } } } @@ -2079,6 +2261,7 @@ impl Drop for Actor { fn drop(&mut self) { self.deleting.abandon_all(); self.take_stalled(); + self.leave_failing("its actor exited"); #[cfg(test)] LIVE_ACTORS.fetch_sub(1, Ordering::SeqCst); } @@ -2142,11 +2325,34 @@ fn publish_indexed(catalog: &ObjectCatalog) { METRIC_OBJECT_STORE_WAL_INDEXED_BYTES.set(bytes as i64); } -/// Returns true for a storage error that a later attempt may not meet: one -/// the object store reports as temporary, or as persistent, which is how its -/// retry layer reports a temporary error that outlasted its retries. -fn is_transient(error: &Error) -> bool { - matches!(error, Error::WalObjectStore { error, .. } if !error.is_permanent()) +/// Returns true for an object store error that repeating the request cannot +/// settle until someone changes the credentials, the configuration or the +/// bucket: denied access; an invalid configuration, which is how OpenDAL +/// reports a missing S3 bucket; and a write to a location that does not +/// exist, which is how OSS, GCS and Azure Blob Storage report a missing +/// bucket or container. A read that finds nothing is not one: it is what the +/// read that reconciles a create meets when the object is gone, and a repeat +/// creates it. Every other error, an HTTP 400 or 5xx, a timeout or a refused +/// connection, may pass by itself. A create that fails with an error +/// requiring intervention is still repeated, with the backoff [`Failing`] +/// describes. +/// +/// Only the kind of the error decides. Its status cannot: on object storage +/// the store sits behind OpenDAL's retry layer, which marks every error it +/// returns as persistent, including those it never retried because they were +/// not temporary, so a 403 arrives with the status of a 503 that outlasted +/// the retries. +fn requires_intervention(error: &Error) -> bool { + match error { + Error::WalObjectStore { + operation, error, .. + } => match error.kind() { + ErrorKind::PermissionDenied | ErrorKind::ConfigInvalid => true, + ErrorKind::NotFound => *operation == "write", + _ => false, + }, + _ => false, + } } /// Wraps an error that several callers receive. @@ -2762,13 +2968,16 @@ mod tests { use common_base::readable_size::ReadableSize; use common_error::ext::{ErrorExt, RetryHint}; + use common_error::status_code::StatusCode; use futures::FutureExt; + use object_store::layers::mock::{self, MockLayerBuilder, oio}; use object_store::services::Memory; + use object_store::util::with_retry_layers; use store_api::logstore::entry::{MultiplePartEntry, MultiplePartHeader}; use tokio::time::timeout; use super::*; - use crate::error::WalObjectStoreSnafu; + use crate::error::{WalObjectConflictSnafu, WalObjectStoreSnafu}; use crate::object_store_wal::batch::{POSITION_LIMIT, entry_id}; use crate::object_store_wal::format::{FOOTER_ENTRY_LEN, decode_object}; @@ -2973,6 +3182,110 @@ mod tests { store.seal_open_batch().await.unwrap(); } + /// An in-memory object store behind the retry layer object storage has + /// in production, whose writes and reads fail with the error kind their + /// switch holds, recording when each refused write started. The retry + /// layer reports such an error as persistent, as it reports a 403 of S3. + struct RefusingStore { + store: ObjectStore, + writes: Arc>>, + reads: Arc>>, + refused_writes: Arc>>, + } + + impl RefusingStore { + fn new() -> Self { + let writes = Arc::new(Mutex::new(None::)); + let reads = Arc::new(Mutex::new(None::)); + let refused_writes = Arc::new(Mutex::new(Vec::new())); + let (refusing_writes, refused_reads) = (writes.clone(), reads.clone()); + let refused_at = refused_writes.clone(); + let mock = MockLayerBuilder::default() + .writer_factory(Arc::new(move |_, _, inner| { + match *refusing_writes.lock().unwrap() { + Some(kind) => { + refused_at.lock().unwrap().push(Instant::now()); + Box::new(RefusedWriter(kind)) + } + None => inner, + } + })) + .reader_factory(Arc::new(move |_, _, inner| { + match *refused_reads.lock().unwrap() { + Some(kind) => Box::new(RefusedReader(kind)), + None => inner, + } + })) + .build() + .unwrap(); + Self { + store: with_retry_layers(memory_store().layer(mock)), + writes, + reads, + refused_writes, + } + } + + fn refuse_writes(&self, kind: Option) { + *self.writes.lock().unwrap() = kind; + } + + fn refuse_reads(&self, kind: Option) { + *self.reads.lock().unwrap() = kind; + } + } + + struct RefusedWriter(ErrorKind); + + impl oio::Write for RefusedWriter { + async fn write(&mut self, _buffer: mock::Buffer) -> mock::Result<()> { + Ok(()) + } + + async fn close(&mut self) -> mock::Result { + Err(mock::Error::new(self.0, "refused by the object store")) + } + + async fn abort(&mut self) -> mock::Result<()> { + Ok(()) + } + } + + struct RefusedReader(ErrorKind); + + impl oio::Read for RefusedReader { + async fn open( + &self, + _range: mock::BytesRange, + ) -> mock::Result<(mock::RpRead, Box)> { + Err(mock::Error::new(self.0, "refused by the object store")) + } + + async fn read( + &self, + _range: mock::BytesRange, + ) -> mock::Result<(mock::RpRead, mock::Buffer)> { + Err(mock::Error::new(self.0, "refused by the object store")) + } + } + + /// Asserts that an append was rejected because the object store refuses + /// the backlog with `kind`: the error is retryable and carries the + /// refusal. + fn assert_rejected(error: &Error, kind: ErrorKind) { + assert!( + matches!(error, Error::ObjectStoreWalFailing { source, .. } + if matches!(&**source, Error::WalObjectStore { error, .. } if error.kind() == kind)), + "unexpected error: {error:?}" + ); + assert_eq!(RetryHint::Retryable, error.retry_hint()); + assert_eq!(StatusCode::StorageUnavailable, error.status_code()); + } + + fn failing_stores() -> i64 { + METRIC_OBJECT_STORE_WAL_FAILING.get() + } + #[tokio::test] async fn test_store_rejects_invalid_config() { let _serialized = SERIALIZED.lock().await; @@ -5624,26 +5937,36 @@ mod tests { let (store, _, mut parked) = open_parking_creates(memory_store(), &eager()).await; let region_id = region(1); let admitted = || *store.admitted_appends.borrow(); + let stalled = METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.get(); + let stalled_waits = METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS.get_sample_count(); - // Every create is parked: once no two more sealed batches fit, - // further appends wait in the channel and are not admitted. + // Every create is parked: once no two more sealed batches fit, the + // next append is held back and counted as stalled, and further + // appends wait in the channel; none is admitted. let mut appends = spawn_appends(&store, region_id, MAX_SEALED_BATCHES - 1).await; for index in 0..3 { let entries = vec![entry(&store, region_id, &format!("w{index}"))]; appends.push(spawn_append_batch(&store, entries)); } let mut releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await; + wait_until(|| METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.get() == stalled + 1).await; round_trip_actor(&store).await; assert_eq!(MAX_SEALED_BATCHES - 1, admitted()); - // A durable object frees a place, and one more append is admitted. + // A durable object frees a place, and the held-back append is + // admitted; the next one is held back in its turn. releases.remove(&1).unwrap().send(true).unwrap(); timeout(WAIT, store.wait_for_admitted_appends(MAX_SEALED_BATCHES)) .await .unwrap() .unwrap(); + wait_until(|| METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.get() == stalled + 2).await; round_trip_actor(&store).await; assert_eq!(MAX_SEALED_BATCHES, admitted()); + assert_eq!( + stalled_waits + 1, + METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS.get_sample_count() + ); // Stop is handled while appends are held back: the batches whose create // has not started and the appends still waiting learn of the stop. @@ -6533,37 +6856,388 @@ mod tests { ); // An object store with a retry layer reports a temporary error that - // outlasted its retries as persistent; the create is repeated too. + // outlasted its retries as persistent; the create is repeated too, + // and appends are still admitted meanwhile. + let rejected = METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.get(); io.fail_next_put_persistently(); append(&store, region_id, "a2").await.unwrap(); - timeout(WAIT, store.wait_durable(&provider(region_id), id(2, 1))) + append(&store, region_id, "a3").await.unwrap(); + timeout(WAIT, store.wait_durable(&provider(region_id), id(3, 1))) .await .unwrap() .unwrap(); - assert_eq!(vec![1, 2], collected_object_seqs(&store).await); + assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await); + assert_eq!( + rejected, + METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.get() + ); } #[tokio::test] - async fn test_store_enqueued_permanent_failure_poisons() { + async fn test_store_classifies_object_store_errors_by_kind() { + let refusing = RefusingStore::new(); + let io = ObjectStoreIo::new(refusing.store.clone(), PREFIX).unwrap(); + let content = Bytes::from_static(b"wal"); + // Behind the retry layer every refusal is persistent, whatever its + // kind, so only the kind tells what a repeat can settle: denied + // access, a missing S3 bucket, or a write to a missing bucket of + // another service cannot; an HTTP 400 or a 5xx that outlasted the + // retries may pass. + for (kind, requires) in [ + (ErrorKind::PermissionDenied, true), + (ErrorKind::ConfigInvalid, true), + (ErrorKind::NotFound, true), + (ErrorKind::Unexpected, false), + ] { + refusing.refuse_writes(Some(kind)); + let error = io.put_if_absent(1, content.clone()).await.unwrap_err(); + assert!( + matches!(&error, Error::WalObjectStore { operation: "write", error, .. } + if error.kind() == kind && error.is_persistent()), + "unexpected error for {kind:?}: {error:?}" + ); + assert_eq!(requires, requires_intervention(&error), "{kind:?}"); + } + + // The object exists, so the create is refused by its condition and + // the read that reconciles it decides: denied access cannot pass, an + // object that is gone is created by the repeat. + refusing.refuse_writes(None); + io.put_if_absent(1, content.clone()).await.unwrap(); + for (kind, requires) in [ + (ErrorKind::PermissionDenied, true), + (ErrorKind::NotFound, false), + ] { + refusing.refuse_reads(Some(kind)); + let error = io + .put_if_absent(1, Bytes::from_static(b"other")) + .await + .unwrap_err(); + assert!( + matches!(&error, Error::WalObjectStore { operation: "read", error, .. } + if error.kind() == kind && error.is_persistent()), + "unexpected error for {kind:?}: {error:?}" + ); + assert_eq!(requires, requires_intervention(&error), "{kind:?}"); + } + assert!(!requires_intervention( + &WalObjectConflictSnafu { + path: io.object_path(1) + } + .build() + )); + } + + #[test] + fn test_store_failing_delay_doubles_up_to_the_cap() { + let now = Instant::now(); + let mut failing = Failing { + error: Arc::new(ObjectStoreWalStoppedSnafu.build()), + since: now, + delay: CREATE_RETRY_DELAY, + probe_at: None, + failed_creates: 1, + rejected_appends: 0, + logged_at: now, + }; + let mut delays = Vec::new(); + for _ in 0..12 { + let delay = failing.delay; + let scheduled_at = Instant::now(); + failing.schedule(); + assert!(failing.probe_at.unwrap() >= scheduled_at + delay); + delays.push(delay.as_millis()); + } + assert_eq!( + vec![ + 100, 200, 400, 800, 1600, 3200, 6400, 12800, 25600, 30000, 30000, 30000 + ], + delays + ); + } + + /// Opens an `enqueued` store over a refusing object store, makes it + /// refuse writes with `kind` and appends one entry, whose create fails, + /// and waits until the store is failing. + async fn open_failing(kind: ErrorKind) -> (Arc, RefusingStore) { + let refusing = RefusingStore::new(); + let store = open(refusing.store.clone(), &enqueued(eager())).await; + let region_id = region(1); + refusing.refuse_writes(Some(kind)); + let response = append(&store, region_id, "a1").await.unwrap(); + assert_eq!( + HashMap::from([(region_id, id(1, 1))]), + response.last_entry_ids + ); + wait_until(|| failing_stores() == 1).await; + (store, refusing) + } + + #[tokio::test] + async fn test_store_enqueued_failure_requiring_intervention_keeps_the_backlog() { + let _serialized = SERIALIZED.lock().await; + let region_id = region(1); + // Denied access and a missing bucket, as S3 reports them. + for kind in [ErrorKind::PermissionDenied, ErrorKind::ConfigInvalid] { + let rejected = METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.get(); + let (store, refusing) = open_failing(kind).await; + let admitted = *store.admitted_appends.borrow(); + + // A later append is rejected at once and takes no entry id; the + // acknowledged entry waits in the backlog and the store is not + // poisoned. + let error = append(&store, region_id, "a2").await.unwrap_err(); + assert_rejected(&error, kind); + assert_eq!(admitted, *store.admitted_appends.borrow()); + assert_eq!( + rejected + 1, + METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.get() + ); + let wait = { + let store = store.clone(); + tokio::spawn( + async move { store.wait_durable(&provider(region_id), id(1, 1)).await }, + ) + }; + timeout(WAIT, store.wait_for_durability_waits(1)) + .await + .unwrap() + .unwrap(); + assert!(!wait.is_finished()); + assert!(terminal(&store.terminal_error).is_none()); + assert_eq!(0, latest(&store, region_id)); + assert_eq!(vec![0], object_seqs(store.io.as_ref()).await); + + // Once the object store accepts writes again, the backlog is + // uploaded under its sequence and admission resumes. + refusing.refuse_writes(None); + timeout(WAIT, wait).await.unwrap().unwrap().unwrap(); + assert_eq!(0, failing_stores()); + assert_eq!( + expected_entries(region_id, &[(id(1, 1), "a1")]), + read_entries(&store, region_id, 1).await + ); + let response = append(&store, region_id, "a3").await.unwrap(); + assert_eq!( + HashMap::from([(region_id, id(2, 1))]), + response.last_entry_ids + ); + timeout(WAIT, store.wait_durable(&provider(region_id), id(2, 1))) + .await + .unwrap() + .unwrap(); + assert_eq!(vec![1, 2], collected_object_seqs(&store).await); + store.stop().await.unwrap(); + } + } + + #[tokio::test] + async fn test_store_enqueued_create_retries_back_off() { + let _serialized = SERIALIZED.lock().await; + let (store, refusing) = open_failing(ErrorKind::PermissionDenied).await; + let refused = || refusing.refused_writes.lock().unwrap().clone(); + wait_until(|| refused().len() >= 5).await; + + // Each attempt starts at least the doubling delay after the previous + // one started, so the 1.6 s from the first refusal hold at most five + // attempts, where a repeat every 100 ms would make sixteen. + let refused = refused(); + for (index, delay) in [100, 200, 400, 800].into_iter().enumerate() { + let gap = refused[index + 1] - refused[index]; + assert!( + gap >= Duration::from_millis(delay), + "attempt {} started {gap:?} after the one before", + index + 1 + ); + } + let window = refused[0] + Duration::from_millis(1600); + assert!(refused.iter().filter(|at| **at < window).count() <= 5); + assert_eq!(1, failing_stores()); + + refusing.refuse_writes(None); + timeout(WAIT, store.wait_durable(&provider(region(1)), id(1, 1))) + .await + .unwrap() + .unwrap(); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_enqueued_failing_store_attempts_the_lowest_batch_alone() { + let _serialized = SERIALIZED.lock().await; + let (store, io, mut parked) = + open_parking_creates(memory_store(), &enqueued(eager())).await; + let region_id = region(1); + for append in spawn_appends(&store, region_id, 3).await { + timeout(WAIT, append).await.unwrap().unwrap().unwrap(); + } + let mut releases = parked_operations(&mut parked, 3).await; + io.creates_refused.store(true, Ordering::SeqCst); + for object_seq in [2, 1, 3] { + releases.remove(&object_seq).unwrap().send(false).unwrap(); + } + + for _ in 0..2 { + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + assert_eq!(1, io.in_flight.load(Ordering::SeqCst)); + release.send(false).unwrap(); + assert!(parked.try_recv().is_err()); + } + + // The attempt that succeeds ends the failing state: the other + // batches are created together and everything is acknowledged in + // order. + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + release.send(true).unwrap(); + for release in parked_operations(&mut parked, 2).await.into_values() { + release.send(true).unwrap(); + } + timeout(WAIT, store.wait_durable(&provider(region_id), id(3, 1))) + .await + .unwrap() + .unwrap(); + assert_eq!(0, failing_stores()); + assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await); + assert_eq!( + expected_entries( + region_id, + &[(id(1, 1), "a1"), (id(2, 1), "a2"), (id(3, 1), "a3")] + ), + read_entries(&store, region_id, 1).await + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_enqueued_rejects_the_appends_held_back_by_a_refused_backlog() { + let _serialized = SERIALIZED.lock().await; + let (store, io, mut parked) = + open_parking_creates(memory_store(), &enqueued(eager())).await; + let region_id = region(1); + let stalled = METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.get(); + let stalled_waits = METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS.get_sample_count(); + let rejected = METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.get(); + + // The sealed batches are full: the next append is held back and the + // ones after it wait in their channel. + for append in spawn_appends(&store, region_id, MAX_SEALED_BATCHES - 1).await { + timeout(WAIT, append).await.unwrap().unwrap().unwrap(); + } + let held = (0..3) + .map(|index| { + spawn_append_batch(&store, vec![entry(&store, region_id, &format!("h{index}"))]) + }) + .collect::>(); + let mut releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await; + wait_until(|| METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.get() == stalled + 1).await; + assert!(held.iter().all(|append| !append.is_finished())); + + // The object store denies access: every append waiting is rejected + // instead of hanging, and so is a later one. + io.creates_refused.store(true, Ordering::SeqCst); + for release in releases.drain().map(|(_, release)| release) { + release.send(false).unwrap(); + } + for append in held { + let error = timeout(WAIT, append).await.unwrap().unwrap().unwrap_err(); + assert_rejected(&error, ErrorKind::PermissionDenied); + } + assert_rejected( + &append(&store, region_id, "a8").await.unwrap_err(), + ErrorKind::PermissionDenied, + ); + assert_eq!( + rejected + 4, + METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.get() + ); + assert_eq!( + stalled_waits + 1, + METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS.get_sample_count() + ); + + // Once a create succeeds the backlog drains and the next append + // takes the sequence after it. + io.creates_refused.store(false, Ordering::SeqCst); + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + release.send(true).unwrap(); + for _ in 2..MAX_SEALED_BATCHES { + next_parked_operation(&mut parked) + .await + .1 + .send(true) + .unwrap(); + } + let last = id(MAX_SEALED_BATCHES as u64 - 1, 1); + timeout(WAIT, store.wait_durable(&provider(region_id), last)) + .await + .unwrap() + .unwrap(); + let appended = spawn_append_batch(&store, vec![entry(&store, region_id, "a9")]); + next_parked_operation(&mut parked) + .await + .1 + .send(true) + .unwrap(); + let response = timeout(WAIT, appended).await.unwrap().unwrap().unwrap(); + assert_eq!( + HashMap::from([(region_id, id(MAX_SEALED_BATCHES as u64, 1))]), + response.last_entry_ids + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_enqueued_stop_uploads_a_refused_backlog_at_once() { + let _serialized = SERIALIZED.lock().await; + let (store, io, mut parked) = + open_parking_creates(memory_store(), &enqueued(eager())).await; + let region_id = region(1); + for append in spawn_appends(&store, region_id, 3).await { + timeout(WAIT, append).await.unwrap().unwrap().unwrap(); + } + io.creates_refused.store(true, Ordering::SeqCst); + for release in parked_operations(&mut parked, 3).await.into_values() { + release.send(false).unwrap(); + } + wait_until(|| failing_stores() == 1).await; + + // Stop does not wait for the next attempt of the failing store: it + // starts the creates of the whole backlog together, which succeed + // now. + let stop = begin_spawned_stop(&store).await; + let releases = parked_operations(&mut parked, 3).await; + for object_seq in 1..=3 { + assert!(releases.contains_key(&object_seq)); + } + for release in releases.into_values() { + release.send(true).unwrap(); + } + timeout(WAIT, stop).await.unwrap().unwrap().unwrap(); + assert_eq!(0, failing_stores()); + assert_eq!(vec![0, 1, 2, 3], object_seqs(io.as_ref()).await); + assert_eq!(id(3, 1), latest(&store, region_id)); + } + + #[tokio::test] + async fn test_store_enqueued_conflict_poisons() { let _serialized = SERIALIZED.lock().await; // A conflict of any epoch poisons the store without reading the // existing object, since the acknowledged entries cannot move to - // another sequence. So does a permanent storage error. - for foreign_epoch in [Some(0), Some(2), None] { + // another sequence. + for foreign_epoch in [0, 2] { let object_store = memory_store(); let (io, reads) = RecordingIo::over(object_store.clone()); let store = open_over(io.clone(), &enqueued(eager())).await; let region_id = region(1); - match foreign_epoch { - Some(epoch) => put_foreign(&object_store, 1, epoch).await, - None => io.fail_next_put_permanently(), - } + put_foreign(&object_store, 1, foreign_epoch).await; let assert_poisoned = |error: Error| { - let poisoned = match foreign_epoch { - Some(_) => matches!(unwrap_shared(&error), Error::WalObjectConflict { .. }), - None => matches!(unwrap_shared(&error), Error::WalObjectStore { .. }), - }; - assert!(poisoned, "unexpected error: {error:?}"); + assert!( + matches!(unwrap_shared(&error), Error::WalObjectConflict { .. }), + "unexpected error: {error:?}" + ); }; // The append was acknowledged; the failure surfaces afterwards. @@ -6642,8 +7316,8 @@ mod tests { read_entries(&store, region_id, 1).await ); - // A backlog that cannot be uploaded is reported by stop. A transient - // failure, temporary or persistent, drops it; a permanent one also + // A backlog that cannot be uploaded is dropped and reported by stop, + // whether the failure is temporary, persistent or permanent; none // poisons the store. for failure in ["temporary", "persistent", "permanent"] { let (io, _) = RecordingIo::over(memory_store()); @@ -6660,10 +7334,7 @@ mod tests { "unexpected error: {error:?}" ); assert_eq!(vec![0], object_seqs(io.as_ref()).await); - assert_eq!( - failure == "permanent", - store.latest_entry_id(&provider(region_id)).is_err() - ); + assert!(terminal(&store.terminal_error).is_none()); store.stop().await.unwrap(); } } @@ -7382,11 +8053,13 @@ mod tests { /// Object access whose whole-object reads, or conditional creates, and /// on request deletes and range reads, park until the test releases /// them, counting how many are in flight. A release of false fails the - /// operation with a transient error before it reaches the object store. + /// operation with a transient error before it reaches the object store, + /// or a create with denied access once `creates_refused` is set. struct ParkedIo { inner: ObjectStoreIo, parked: mpsc::UnboundedSender<(u64, oneshot::Sender)>, park_creates: bool, + creates_refused: AtomicBool, /// Whether creates park yet; set once the store opened. creates_parked: AtomicBool, /// Whether deletes park. @@ -7428,6 +8101,7 @@ mod tests { inner: ObjectStoreIo::new(object_store, PREFIX).unwrap(), parked, park_creates, + creates_refused: AtomicBool::new(false), creates_parked: AtomicBool::new(false), deletes_parked: AtomicBool::new(false), range_reads_parked: AtomicBool::new(false), @@ -7458,6 +8132,17 @@ mod tests { && self.creates_parked.load(Ordering::SeqCst) && !self.park(object_seq).await { + if self.creates_refused.load(Ordering::SeqCst) { + return Err(object_store::Error::new( + ErrorKind::PermissionDenied, + "access denied", + ) + .set_persistent()) + .context(WalObjectStoreSnafu { + operation: "write", + path: self.object_path(object_seq), + }); + } return injected_failure("write", self.object_path(object_seq)); } self.inner.put_if_absent(object_seq, content).await @@ -7660,8 +8345,10 @@ mod tests { EarlierEpochDurable, /// transient I/O error, `enqueued` mode TransientEnqueued, - /// transient I/O error, `enqueued` mode after `stop` began - TransientEnqueuedAfterStop, + /// I/O error that requires intervention, `enqueued` mode + InterventionEnqueued, + /// any I/O error, `enqueued` mode after `stop` began + ErrorEnqueuedAfterStop, /// an object of the same or a later epoch holds the sequence, or /// any conflicting object in the `enqueued` mode Conflict, @@ -7677,12 +8364,13 @@ mod tests { } /// Every row of the matrix, in the order the RFC lists them. - const ROWS: [Row; 10] = [ + const ROWS: [Row; 11] = [ Row::Created, Row::TransientDurable, Row::EarlierEpochDurable, Row::TransientEnqueued, - Row::TransientEnqueuedAfterStop, + Row::InterventionEnqueued, + Row::ErrorEnqueuedAfterStop, Row::Conflict, Row::CatalogError, Row::LastSequence, @@ -7746,6 +8434,7 @@ mod tests { skipped_segments: METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(), stalled_appends: METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.get(), stalled_waits: METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS.get_sample_count(), + rejected_appends: METRIC_OBJECT_STORE_WAL_REJECTED_APPENDS_TOTAL.get(), } /// Samples of the two object shape histograms, which a sealed batch @@ -7941,11 +8630,21 @@ mod tests { row: Some(Row::TransientEnqueued), run: || Box::pin(create_fails_transiently_enqueued()), }, + Case { + fault: "creates the object store refuses with an error that requires intervention, `enqueued` mode", + row: Some(Row::InterventionEnqueued), + run: || Box::pin(creates_refused_enqueued()), + }, Case { fault: "a create that fails transiently after `stop` began, `enqueued` mode", - row: Some(Row::TransientEnqueuedAfterStop), + row: Some(Row::ErrorEnqueuedAfterStop), run: || Box::pin(create_fails_transiently_after_stop_began()), }, + Case { + fault: "a stop while the object store refuses the backlog, `enqueued` mode", + row: Some(Row::ErrorEnqueuedAfterStop), + run: || Box::pin(stop_while_the_backlog_is_refused()), + }, Case { fault: "a create that finds an object of an earlier epoch, `durable` mode", row: Some(Row::EarlierEpochDurable), @@ -8061,6 +8760,11 @@ mod tests { row: None, run: || Box::pin(create_conflicts_after_stop_began()), }, + Case { + fault: "an append held back by full sealed batches, `durable` mode", + row: None, + run: || Box::pin(append_held_back_by_full_sealed_batches()), + }, Case { fault: "a stalled append released by admission, then one released by a stop", row: None, @@ -8414,6 +9118,157 @@ mod tests { ); } + /// Objects 1 to 3 are acknowledged when the object store starts + /// denying access. The store keeps them under their sequences, + /// rejects the next append and attempts object 1 alone; once it is + /// created the others follow and appends are admitted again. + async fn creates_refused_enqueued() { + let (store, io, mut parked) = + open_parking_creates(memory_store(), &enqueued(eager())).await; + let before = Counters::sample(); + let region_id = region(1); + for append in spawn_appends(&store, region_id, 3).await { + timeout(WAIT, append).await.unwrap().unwrap().unwrap(); + } + let mut releases = parked_operations(&mut parked, 3).await; + io.creates_refused.store(true, Ordering::SeqCst); + for object_seq in 1..=3 { + releases.remove(&object_seq).unwrap().send(false).unwrap(); + } + wait_until(|| Counters::since(before).create_failures == 3).await; + assert_eq!(1, failing_stores()); + let error = append(&store, region_id, "a4").await.unwrap_err(); + assert_rejected(&error, ErrorKind::PermissionDenied); + assert_eq!(0, store.durable_entry_id(&provider(region_id)).unwrap()); + assert!(read_entries(&store, region_id, 1).await.is_empty()); + assert!(terminal(&store.terminal_error).is_none()); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + create_failures: 3, + sealed_objects: 3, + object_bytes: 3 * one_entry_object(), + object_entries: 3, + reads: 1, + rejected_appends: 1, + ..Counters::default() + }, + Counters::since(before) + ); + + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + assert!(parked.try_recv().is_err()); + release.send(true).unwrap(); + for release in parked_operations(&mut parked, 2).await.into_values() { + release.send(true).unwrap(); + } + timeout(WAIT, store.wait_durable(&provider(region_id), id(3, 1))) + .await + .unwrap() + .unwrap(); + assert_eq!(0, failing_stores()); + let response = append(&store, region_id, "a5").await.unwrap(); + assert_eq!(Some(&id(4, 1)), response.last_entry_ids.get(®ion_id)); + next_parked_operation(&mut parked) + .await + .1 + .send(true) + .unwrap(); + timeout(WAIT, store.wait_durable(&provider(region_id), id(4, 1))) + .await + .unwrap() + .unwrap(); + assert_eq!(vec![1, 2, 3, 4], collected_object_seqs(&store).await); + assert_eq!( + expected_entries( + region_id, + &[ + (id(1, 1), "a1"), + (id(2, 1), "a2"), + (id(3, 1), "a3"), + (id(4, 1), "a5") + ] + ), + read_entries(&store, region_id, 1).await + ); + assert_indexed(io.as_ref(), &[1, 2, 3, 4]).await; + assert_eq!( + Counters { + created_objects: 4, + seal_to_durable: 4, + create_failures: 3, + sealed_objects: 4, + object_bytes: 4 * one_entry_object(), + object_entries: 4, + reads: 2, + deleted_objects: 1, + rejected_appends: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// The object store denies access to objects 1 and 2 and still does + /// when `stop` begins: stop attempts both together, drops the backlog + /// when they fail and reports the refusal; the store is not poisoned. + async fn stop_while_the_backlog_is_refused() { + let (store, io, mut parked) = + open_parking_creates(memory_store(), &enqueued(eager())).await; + let before = Counters::sample(); + let region_id = region(1); + for append in spawn_appends(&store, region_id, 2).await { + timeout(WAIT, append).await.unwrap().unwrap().unwrap(); + } + io.creates_refused.store(true, Ordering::SeqCst); + for release in parked_operations(&mut parked, 2).await.into_values() { + release.send(false).unwrap(); + } + wait_until(|| Counters::since(before).create_failures == 2).await; + assert_eq!(1, failing_stores()); + let wait = { + let store = store.clone(); + tokio::spawn( + async move { store.wait_durable(&provider(region_id), id(1, 1)).await }, + ) + }; + timeout(WAIT, store.wait_for_durability_waits(1)) + .await + .unwrap() + .unwrap(); + + let stop = { + let store = store.clone(); + tokio::spawn(async move { store.stop().await }) + }; + for release in parked_operations(&mut parked, 2).await.into_values() { + release.send(false).unwrap(); + } + let error = timeout(WAIT, stop).await.unwrap().unwrap().unwrap_err(); + assert!( + matches!(unwrap_shared(&error), Error::WalObjectStore { error, .. } + if error.kind() == ErrorKind::PermissionDenied), + "unexpected error: {error:?}" + ); + assert_stopped(&timeout(WAIT, wait).await.unwrap().unwrap().unwrap_err()); + assert!(terminal(&store.terminal_error).is_none()); + assert_eq!(0, failing_stores()); + assert_eq!(vec![0], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + create_failures: 4, + sealed_objects: 2, + object_bytes: 2 * one_entry_object(), + object_entries: 2, + ..Counters::default() + }, + Counters::since(before) + ); + } + async fn create_finds_an_earlier_epoch() { let object_store = memory_store(); let store = open(object_store.clone(), &eager()).await; @@ -9581,6 +10436,63 @@ mod tests { ); } + /// Seven sealed batches leave no room for the two an admission can + /// seal: the next append is held back and counted as stalled, and it + /// is admitted once object 1 is durable, which records its wait. + async fn append_held_back_by_full_sealed_batches() { + let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await; + let before = Counters::sample(); + let region_id = region(1); + let mut appends = spawn_appends(&store, region_id, MAX_SEALED_BATCHES - 1).await; + appends.push(spawn_append_batch( + &store, + vec![entry(&store, region_id, "h1")], + )); + let releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await; + wait_until(|| Counters::since(before).stalled_appends == 1).await; + assert_eq!(0, Counters::since(before).stalled_waits); + + for release in releases.into_values() { + release.send(true).unwrap(); + } + for _ in MAX_IN_FLIGHT_CREATES..MAX_SEALED_BATCHES { + next_parked_operation(&mut parked) + .await + .1 + .send(true) + .unwrap(); + } + for (index, append) in appends.into_iter().enumerate() { + let response = timeout(WAIT, append).await.unwrap().unwrap().unwrap(); + assert_eq!( + Some(&id(index as u64 + 1, 1)), + response.last_entry_ids.get(®ion_id) + ); + } + let sealed = MAX_SEALED_BATCHES as u64; + assert_eq!( + (1..=sealed).collect::>(), + collected_object_seqs(&store).await + ); + assert_indexed(io.as_ref(), &(1..=sealed).collect::>()).await; + assert_eq!( + Counters { + created_objects: sealed, + seal_to_durable: sealed, + sealed_objects: sealed, + object_bytes: sealed * one_entry_object(), + object_entries: sealed, + acknowledged_appends: sealed, + deleted_objects: 1, + stalled_appends: 1, + stalled_waits: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + /// The `enqueued` configuration under which the second append stalls /// behind the create of the first. fn stalling() -> ObjectStoreWalConfig { diff --git a/src/mito2/src/engine/object_store_wal_recovery_test.rs b/src/mito2/src/engine/object_store_wal_recovery_test.rs index 126e38e4b26..54626b24432 100644 --- a/src/mito2/src/engine/object_store_wal_recovery_test.rs +++ b/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) -> 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 { + 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 + ); +}