diff --git a/Cargo.lock b/Cargo.lock index b0c66cc8b5c..edf460e4679 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8591,6 +8591,7 @@ dependencies = [ "humantime-serde", "itertools 0.14.0", "lazy_static", + "log-store", "mito-codec", "mito2", "moka", diff --git a/docs/rfcs/2026-09-06-object-store-wal.md b/docs/rfcs/2026-09-06-object-store-wal.md index b49f97c330c..1d784dd09ae 100644 --- a/docs/rfcs/2026-09-06-object-store-wal.md +++ b/docs/rfcs/2026-09-06-object-store-wal.md @@ -31,7 +31,7 @@ Object storage is already provisioned, already durable across zones, and already # Non-goals - Distributed mode. The provider is rejected when a datanode is controlled by a metasrv; the metasrv side has no allocation logic for it yet. -- Persistence of per-region obsolete watermarks. Mito re-establishes every watermark when it opens a region, which is what garbage collection reasons from; persisting them is a follow-up. +- Persistence of the obsolete watermarks that flushes and truncates move. Mito re-establishes each of them when it opens a region, which is what garbage collection reasons from. Only the release of a region that is never opened again, after a drop or an offline cleanup, is persisted, as described under *Entry ids*. - Matching Raft Engine's write latency. The provider trades latency for the absence of local durable WAL state. - Making standalone metadata independent of local disk. The metadata store is unchanged by this RFC. @@ -65,11 +65,12 @@ header | segment(region 1) | ... | segment(region N) | footer | trailer - The header carries the magic `GTWALOBJ`, format version 1, the object sequence number, the epoch of the store instance that wrote the object, a link to the object this one extends, as that object's sequence number and epoch or none for the first object of a chain, the reclaim boundary described under *Entry ids*, and a CRC32 of the preceding header bytes; the header is 54 bytes. A reclaim boundary above the object's own sequence fails the header check. A sequence number and an epoch together identify an object: an instance never writes two objects under one sequence, and instances have distinct epochs, as described under *Log store*. Recovery selects the history it replays by the links and the epochs, as described under *Log store*. - An object may hold no segment. A store writes such an object when it opens, as described under *Log store*; it is a chain object like any other, and its segments are trivially all flushed. +- An object may carry release records, which a store writes when it releases a region, as described under *Entry ids*; each names a region and the entry id the region is released through. They form the single entry of a segment of the reserved region id `u64::MAX`, whose entry id is at position 1 of the object; segments are ordered by region id, so it is the last segment of the object. Its payload is their count, then per record, ordered by region id and at most one per region, the region id and the entry id. The object keeps format version 1, so a reader that predates release records reads it as any other object. An object written for a release alone holds only that segment. - Each segment holds the entries of exactly one region, ordered by entry id; segments are ordered by region id. Ordering makes the encoding deterministic, which the identical-retry rule below depends on. - The footer records, per segment, the region id, the entry id range, the entry count, the byte range and a CRC32 of the segment. - The fixed-size trailer carries the magic `GTWALTRL`, the footer location, a CRC32 of the footer and a CRC32 of the whole object. A reader locates the footer by reading the trailer at the end of the object and can decode a single region's segment without touching the others. -Corruption is handled at two levels; nothing else is checked in production. Structural damage fails recovery: recovery reads only the header, the trailer and the footer of every object, as described under *Log store*, and verifies the header and trailer magic, the format version, the header CRC, the header sequence against the key, the footer CRC and that the segment byte ranges tile the object body exactly; the first object that fails any of these checks fails construction of the store, since nothing can be trusted about such an object. A segment is verified by the read that decodes it, and a segment that does not decode, because its CRC does not match or its content disagrees with its footer entry, is a data-level failure confined to one region. The read fetches the segment once more, since the download rather than the object may be damaged, and decodes it again; if it still does not decode, the behaviour follows `on_corrupted_segment`. `skip`, the default, skips the segment, records a WAL hole for the region with the object key, the object sequence and the entry id range the footer holds for the segment, increments the counter `greptime_logstore_object_store_wal_skipped_segments_total` and logs a warning with those fields, and the read continues with the next object; every other region in the object is unaffected. The holes are kept in memory per region and can be listed from the store; nothing is persisted, so a store opened later on the same prefix records a hole again when a read meets the segment. `fail` fails the read with the decode error, so a region whose replay meets the segment does not open. Skipping loses only the unflushed entries of that region in that object: nothing already flushed is affected, and the loss is reported rather than hidden. A failed fetch is not corruption and fails the read in both modes, unless garbage collection deleted the object, as described under *Entry ids*, in which case the object is skipped; an object the catalog still holds once a delete of it has settled fails the read whether it is missing or unreadable, and a segment of such an object that does not decode follows `on_corrupted_segment` as it would without any collection. The whole-object CRC is not a recovery requirement; it stays in the trailer as a fixture-level check. Because recovery never checks it, the header carries its own CRC, so a damaged link fails recovery instead of silently cutting acknowledged objects off the chain. The format is frozen at version 1. +Corruption is handled at two levels; nothing else is checked in production. Structural damage fails recovery: recovery reads only the header, the trailer and the footer of every object, as described under *Log store*, and verifies the header and trailer magic, the format version, the header CRC, the header sequence against the key, the footer CRC and that the segment byte ranges tile the object body exactly; the first object that fails any of these checks fails construction of the store, since nothing can be trusted about such an object. The segment of release records of an indexed object, which recovery reads too, is held to the same rule rather than to `on_corrupted_segment`: one that cannot be fetched or does not decode fails construction, since the watermarks it seeds would otherwise be lost. A segment is verified by the read that decodes it, and a segment that does not decode, because its CRC does not match or its content disagrees with its footer entry, is a data-level failure confined to one region. The read fetches the segment once more, since the download rather than the object may be damaged, and decodes it again; if it still does not decode, the behaviour follows `on_corrupted_segment`. `skip`, the default, skips the segment, records a WAL hole for the region with the object key, the object sequence and the entry id range the footer holds for the segment, increments the counter `greptime_logstore_object_store_wal_skipped_segments_total` and logs a warning with those fields, and the read continues with the next object; every other region in the object is unaffected. The holes are kept in memory per region and can be listed from the store; nothing is persisted, so a store opened later on the same prefix records a hole again when a read meets the segment. `fail` fails the read with the decode error, so a region whose replay meets the segment does not open. Skipping loses only the unflushed entries of that region in that object: nothing already flushed is affected, and the loss is reported rather than hidden. A failed fetch is not corruption and fails the read in both modes, unless garbage collection deleted the object, as described under *Entry ids*, in which case the object is skipped; an object the catalog still holds once a delete of it has settled fails the read whether it is missing or unreadable, and a segment of such an object that does not decode follows `on_corrupted_segment` as it would without any collection. The whole-object CRC is not a recovery requirement; it stays in the trailer as a fixture-level check. Because recovery never checks it, the header carries its own CRC, so a damaged link fails recovery instead of silently cutting acknowledged objects off the chain. The operator configures only the prefix root. The store runs under a node prefix derived from it, `/datanodes//epochs/`, and object keys are `/objects/.wal`, where the sequence is zero-padded to 20 decimal digits. Standalone uses its datanode id, zero unless configured, and generation zero, so with prefix `wal` the first object is `wal/datanodes/0/epochs/0/objects/00000000000000000000.wal`. Every place that names the prefix, whether to open the store or to allocate the WAL options of a region, derives it from the same function, so the prefix a region persists matches the one its store runs under. Listing accepts only well-formed keys under `/objects/` and returns them in sequence order. This is the standalone form of the key layout under *Alignment with cluster mode*; the cluster assignment of node id and generation is proposed there. @@ -105,17 +106,19 @@ Two rules keep every new id of a region above every id it already has, on a pref The second rule is the sequence floor of `obsolete`. Mito calls `obsolete` with the manifest's `flushed_entry_id` when it opens a region, and the actor applies the watermark and a floor derived from that entry id together: the next object sequence is raised to `(entry_id >> 20) + 1` unless it is there already, while zero, the watermark of a region without entries, names no object and needs no floor, so that a watermark assigned under another prefix, or one whose object this prefix no longer holds, is never passed by a new id. The floor moves only the next sequence, never the chain: the next batch still links to its predecessor, so a skipped sequence leaves no gap in the history recovery replays. While the open batch holds ids under the next sequence, the call is refused with a retryable error and records neither the watermark nor the floor, because those ids would fall below the floor. In the `enqueued` mode an id the store itself handed out needs no floor, even before it is durable: it is never handed out again while the store runs, so every later id of the region is greater. In the current single-prefix layout the floor never needs to move, since every watermark Mito hands the store names an object below the next sequence; the rule exists for the cross-prefix cases under *Alignment with cluster mode*. No additional persisted state is introduced. -Garbage collection deletes the objects below a reclaim boundary that the store persists in every header, so that recovery can tell a collected predecessor from one that never landed. The boundary is computed from the catalog and the watermarks the store already keeps: each region with a recorded watermark contributes the sequence of its first indexed object holding a segment whose largest entry id is above the watermark; a region without a recorded watermark contributes its first indexed object; a region whose watermark is the maximum entry id, after `obsolete_all`, contributes nothing; the boundary is the smallest contribution, or the last indexed sequence plus one when nothing contributes, and it never decreases. Whether a segment is flushed is decided per segment from the footer, on the full entry id, so a watermark that stops in the middle of an object keeps the object. The boundary is recomputed after every successful `obsolete`, after `obsolete_all` and after every object is indexed; Mito calls `obsolete` after a flush, after a truncate, when it drops a region and when it opens one, so a restart re-establishes every watermark and the boundary advances on the same rule. Every sealed batch carries the boundary current at sealing and the start object carries the boundary recovery found, so a boundary is always at or below the sequence of the object carrying it. The boundary that permits deletion is the one carried by the last indexed object; the boundary of a batch whose create failed or is in flight permits nothing. +Garbage collection deletes the objects below a reclaim boundary that the store persists in every header, so that recovery can tell a collected predecessor from one that never landed. The boundary is computed from the catalog and the watermarks the store already keeps: each region with a recorded watermark contributes the sequence of its first indexed object holding a segment whose largest entry id is above the watermark; a region without a recorded watermark contributes its first indexed object; a region whose watermark is the maximum entry id, after a release in full, contributes nothing; the boundary is the smallest contribution, or the last indexed sequence plus one when nothing contributes, and it never decreases. Whether a segment is flushed is decided per segment from the footer, on the full entry id, so a watermark that stops in the middle of an object keeps the object. The boundary is recomputed after every successful `obsolete`, after every release and after every object is indexed; Mito calls `obsolete` after a flush, after a truncate, when it drops a region and when it opens one, so a restart re-establishes the watermark of every region Mito opens again, and the boundary advances on the same rule. A region that is never opened again needs the release described below. Every sealed batch carries the boundary current at sealing and the start object carries the boundary recovery found, so a boundary is always at or below the sequence of the object carrying it. The boundary that permits deletion is the one carried by the last indexed object; the boundary of a batch whose create failed or is in flight permits nothing. -Every present object with a sequence below the durable boundary is deletable, on the chain or not: the objects the catalog indexes, the objects recovery listed but did not index, the objects of an earlier epoch that the start object moved past, and the sequences of batches that failed in this run, whose objects may or may not exist. A collection starts after a successful `obsolete`, after `obsolete_all` and after an indexing that raised the durable boundary. Deletes run in the background, at most four at a time and in no particular order, and never hold up admission, sealing, uploads or acknowledgements; an indexed object is unindexed only after its delete succeeded, and the WAL holes recorded for it are dropped with it. Every deleted object is counted in `greptime_logstore_object_store_wal_deleted_objects_total`; a failed delete is logged, counted in `greptime_logstore_object_store_wal_failed_deletes_total` and retried at the next collection, and the object stays indexed meanwhile. An object deleted but not yet unindexed when the process crashes is simply absent at the next recovery, which lists the prefix. Nothing is collected once `stop` has begun or while the store is poisoned; a delete in flight when `stop` begins runs to completion before `stop` returns, and the actor settles every attempt it abandons when it exits. +Every present object with a sequence below the durable boundary is deletable, on the chain or not: the objects the catalog indexes, the objects recovery listed but did not index, the objects of an earlier epoch that the start object moved past, and the sequences of batches that failed in this run, whose objects may or may not exist. A collection starts after a successful `obsolete`, after a release and after an indexing that raised the durable boundary. Deletes run in the background, at most four at a time and in no particular order, and never hold up admission, sealing, uploads or acknowledgements; an indexed object is unindexed only after its delete succeeded, and the WAL holes recorded for it are dropped with it. Every deleted object is counted in `greptime_logstore_object_store_wal_deleted_objects_total`; a failed delete is logged, counted in `greptime_logstore_object_store_wal_failed_deletes_total` and retried at the next collection, and the object stays indexed meanwhile. An object deleted but not yet unindexed when the process crashes is simply absent at the next recovery, which lists the prefix. Nothing is collected once `stop` has begun or while the store is poisoned; a delete in flight when `stop` begins runs to completion before `stop` returns, and the actor settles every attempt it abandons when it exits. -A read never needs a deleted object, since entries at or below a region's watermark are hidden anyway, but a read that listed an object before it was collected skips the object when it gets there. A read that meets an error on such an object makes two observations, in the reverse of the order the actor writes them: it looks at the set of objects whose delete is in flight, and then at the catalog. The actor registers an object in that set before its delete starts, unindexes it if the delete succeeded, and only then takes it out, so an object absent from the set was either never collected, in which case the catalog still holds it, or is already unindexed by the time the catalog is read. Observing the catalog first would let a collection complete between the two observations and leave the read with an object that still looked indexed and no longer looked deleting. An object that is still in the set has an undecided delete, so the read waits for that one delete, a single object store request, rather than assume it succeeds: a delete that fails leaves the object present and indexed, and the error the read met is then reported as any other, `on_corrupted_segment` included. What the read waits for is the delete attempt its observation found, taken together with the lookup, not the object: a retry of the same object is a separate attempt, so the attempt the read holds releases it even while that retry is in flight, and a read is never handed from one attempt to the next. A read stream outlives the store it came from, so the actor settles every attempt it abandons when it exits, whether the store was stopped or dropped; the objects of those attempts stay present and indexed, so the reads waiting for them report the error they met. A collected object is skipped silently: it is not recorded as a hole and does not count as a skipped segment. An object the catalog still holds after those observations fails the read like any other fetch failure, `on_corrupted_segment` included. Within one run the catalog keeps the largest entry id of a region after the objects holding it were collected, so `latest_entry_id`, the durability wait and the `enqueued` clamp keep their reference point; a store opened later learns only what is listed, so after a restart `latest_entry_id` of a region whose later objects were collected is the largest id among the objects left, which only the pruning hint of Mito consumes. No state is persisted outside the headers and no configuration option is added. +A region that Mito drops is never opened again, so no restart re-establishes its watermark, and a region without a watermark holds the boundary at its first indexed object. The store therefore persists the release of such a region in the WAL. `obsolete_all`, which Mito calls when it cleans a region up offline, releases the region in full, through the maximum entry id. `release_region`, a `LogStore` method that Mito calls after `obsolete` when it drops a region, releases it through the largest entry id the store has handed out for the region, so entries of a region that reuses the id later stay above the release. A release moves the region's watermark up in memory, recomputes the boundary, starts a collection and seals the open batch at once, empty or not, with a release record of the region. The call returns once that object is durable and indexed, in both acknowledgement modes, and fails whenever its batch fails, as an append in the `durable` mode does. In the `enqueued` mode the watermark of a release is not clamped to the durable entry id: a released region is never replayed, and its record becomes durable only after every object sealed before it. The object of a release carries the boundary its watermark allows, so the record and that boundary become durable together. Recovery seeds the watermarks from the release records of the indexed objects before it computes any boundary, and a later `obsolete` never moves a watermark down. A release record needs no bookkeeping of its own. It is sealed once every entry it covers was admitted, so it lies in the object holding the last segment it covers or in a later one; collection deletes, in any order, only objects below the durable boundary; and recovery indexes only objects at or above the boundary of the tip, which every start object carries forward. So recovery indexes a release record whenever it indexes a segment the record covers, and once the record is below the boundary, so is every segment it covers, which recovery never indexes again, whether collection deleted it yet or not. A release in full also covers the entries a region would be assigned afterwards, so it is made only for a region that writes nothing more. When the batch of a release fails, the call fails, and the store carries the record again in the next batch it seals; while such a record waits, a timer tick seals it even when no entry is admitted, and a stop in the `durable` mode gives it one more create and reports the record lost when that create fails, so the release becomes durable once a create succeeds again. While the store is failing, as described under *Log store*, a release fails at once with the retryable error rejected appends carry, and so does a release whose batch is still pending when the store becomes failing; neither poisons the store. The record of the first is kept like that of a failed batch, but no tick seals it while the store is failing, so repeated releases add no batch to the backlog; the first tick after a create succeeds seals it, and a `stop` seals it with the backlog for its last attempt. The record of the second stays in the backlog. Both become durable once a create succeeds. A crash before that loses the record. Two windows therefore remain in which a dropped region is never released: a crash between the deletion of the table metadata and a durable release record, which includes a crash during the drop itself, and a crash after a failed release before any later create succeeds. Nothing retries the drop of such a region. The first window is a gap of the drop procedure rather than of the WAL, issue #9577 (https://github.com/GreptimeTeam/greptimedb/issues/9577): a crash during the drop leaves the region's data behind whatever the WAL provider. In the second window the drop completed except for its release, and a standalone region drop that fails is not retried. An operator sees `greptime_logstore_object_store_wal_indexed_objects` only grow after the restart while `deleted_objects_total` stays put, because the region's segments hold the reclaim boundary. Its remedy belongs in the drop procedure, for example an offline cleanup of a region that a resumed region drop finds not open, whose `obsolete_all` writes a durable release. Reclamation that consults the metadata for live regions, as Kafka's pruning by the metasrv does, would be a second safety net for both windows; it is left to garbage collection across nodes, see *Alignment with cluster mode*. Neither is part of this change. The object of the failed batch is seeded only when nothing was acknowledged after it, as the entries of a failed append are replayed. Every release record in an object the store indexes is counted in `greptime_logstore_object_store_wal_release_records_total`, records carried over from a failed batch included. + +A read never needs a deleted object, since entries at or below a region's watermark are hidden anyway, but a read that listed an object before it was collected skips the object when it gets there. A read that meets an error on such an object makes two observations, in the reverse of the order the actor writes them: it looks at the set of objects whose delete is in flight, and then at the catalog. The actor registers an object in that set before its delete starts, unindexes it if the delete succeeded, and only then takes it out, so an object absent from the set was either never collected, in which case the catalog still holds it, or is already unindexed by the time the catalog is read. Observing the catalog first would let a collection complete between the two observations and leave the read with an object that still looked indexed and no longer looked deleting. An object that is still in the set has an undecided delete, so the read waits for that one delete, a single object store request, rather than assume it succeeds: a delete that fails leaves the object present and indexed, and the error the read met is then reported as any other, `on_corrupted_segment` included. What the read waits for is the delete attempt its observation found, taken together with the lookup, not the object: a retry of the same object is a separate attempt, so the attempt the read holds releases it even while that retry is in flight, and a read is never handed from one attempt to the next. A read stream outlives the store it came from, so the actor settles every attempt it abandons when it exits, whether the store was stopped or dropped; the objects of those attempts stay present and indexed, so the reads waiting for them report the error they met. A collected object is skipped silently: it is not recorded as a hole and does not count as a skipped segment. An object the catalog still holds after those observations fails the read like any other fetch failure, `on_corrupted_segment` included. Within one run the catalog keeps the largest entry id of a region after the objects holding it were collected, so `latest_entry_id`, the durability wait and the `enqueued` clamp keep their reference point; a store opened later learns only what is listed, so after a restart `latest_entry_id` of a region whose later objects were collected is the largest id among the objects left, which only the pruning hint of Mito consumes. Reclaim boundaries are persisted in the headers and release records in the segments of the reserved region, no other state is persisted, and no configuration option is added. ## Log store `ObjectStoreLogStore` implements the `LogStore` trait for `Provider::ObjectStore`. -**Construction and recovery.** The prefix, flush interval, batch size and backlog thresholds are validated, then the store lists the prefix and fetches, for up to eight objects at a time, the header and a window at the end of each object with range reads; a short object is read whole, and the footer is read separately only when it starts before the window. Each header and footer is verified as described under *Object format*; the first corrupted object fails construction; a fetch that finds the object gone is judged once the history is selected, as described below, and any other failed fetch abandons the rest. Segments are not read, so recovery costs a few small reads per object however large the objects are. +**Construction and recovery.** The prefix, flush interval, batch size and backlog thresholds are validated, then the store lists the prefix and fetches, for up to eight objects at a time, the header and a window at the end of each object with range reads; a short object is read whole, and the footer is read separately only when it starts before the window. Each header and footer is verified as described under *Object format*; the first corrupted object fails construction; a fetch that finds the object gone is judged once the history is selected, as described below, and any other failed fetch abandons the rest. Segments are not read, except the segment of release records of each indexed object that holds one, with one range read, so recovery costs a few small reads per object however large the objects are. The store then selects the history it replays from the headers. An object's link holds when the object has no predecessor, when its predecessor is present and carries the epoch the link records, or when the predecessor's sequence is below the reclaim boundary the object carries, in which case the chain ends there whether or not the predecessor is present; above that boundary a missing predecessor breaks the link. An object is complete when every link on its chain holds, each judged with the boundary of the object carrying the link. The tip is the complete object with the largest epoch and, among those, the largest sequence. Only the objects on the tip's chain are indexed and replayed, in sequence order whatever the order the fetches completed in, so the catalog checks the entry ranges of every object against its predecessors like a sequential replay, and a conflicting object fails construction. Every other present object is an orphan: it is neither indexed nor read, but its sequence counts, so the next sequence is above every listed object, and the objects below the tip's boundary are collected. A listed object whose header or footer is gone by the time it is fetched is ignored when it lies below the tip's boundary and fails construction when it lies at or above it, since collection never deletes one there. Construction also fails when a listed object carries an epoch above the next sequence, which no instance can have produced, since an epoch is a start sequence plus one. When objects are present but none is complete, construction fails rather than start an empty history. Construction then records the largest durable entry id per region. @@ -142,28 +145,28 @@ The mode also needs a barrier that the durable mode gets for free. An entry id t | create succeeds, or identical retry | advances | acknowledged in sequence order | healthy | | 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 | -| 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 | +| transient I/O error, `enqueued` mode | unchanged | already acknowledged, except a release, which waits for the repeat | healthy: the create is repeated after a short delay | +| I/O error that requires intervention, `enqueued` mode | unchanged | already acknowledged, except a release, which fails with a retryable error while its record stays in the backlog; later appends and releases are rejected with that 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, except a release, which fails with the stopped error | 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 | | the start object meets an object of an earlier epoch | moves to the next sequence | none admitted yet | construction continues | | the start object meets an object of the same or a later epoch, or its create has an unknown outcome | unchanged | none admitted yet | construction fails; an unknown outcome is retryable | -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. +A release is a waiter of the batch that carries its record in both modes: it returns when the batch is indexed and fails whenever the batch fails. 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, 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. +**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, and releases fail with the stopped error. In the `durable` mode the release records of failed batches and of the batches dropped here are sealed into one more object, which is created before `stop` returns; `stop` reports the error when that create fails too. 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, unless the poisoning drops release records, which `stop` reports by returning the conflict. 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. +**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 the store persists no watermark of a region Mito opens again. The release of a region, which `obsolete_all` and `release_region` make, is persisted as described under *Entry ids*. `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 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. +**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 and releases such a store rejected; the attempts of a failing store are counted in `create_failures_total`. The delete counters of a collection and the release record counter 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. ## Mito integration -Mito maps `WalOptions::ObjectStore { prefix }` to `Provider::object_store_provider(region_id, prefix)`; resolving object store options against a Raft Engine or Kafka log store reports the existing incompatible-provider error. Reads go through the region-scoped reader used for Kafka, so a namespace shared by many regions never leaks entries across regions even if a log store returned them. Object store regions are opened individually rather than grouped by topic. After replay, `topic_latest_entry_id` follows the existing remote-WAL rule: it is taken from the store's latest entry id only when the region's memtables are empty, otherwise it stays at the replay start, because that value feeds pruning and must never advance past unflushed data. Before a flush, a full truncate or a discard of unflushed data publishes a manifest watermark, Mito waits on the durability barrier described under *Acknowledgement*; with the `durable` mode and with every other provider the wait returns at once. +Mito maps `WalOptions::ObjectStore { prefix }` to `Provider::object_store_provider(region_id, prefix)`; resolving object store options against a Raft Engine or Kafka log store reports the existing incompatible-provider error. Reads go through the region-scoped reader used for Kafka, so a namespace shared by many regions never leaks entries across regions even if a log store returned them. Object store regions are opened individually rather than grouped by topic. After replay, `topic_latest_entry_id` follows the existing remote-WAL rule: it is taken from the store's latest entry id only when the region's memtables are empty, otherwise it stays at the replay start, because that value feeds pruning and must never advance past unflushed data. Before a flush, a full truncate or a discard of unflushed data publishes a manifest watermark, Mito waits on the durability barrier described under *Acknowledgement*; with the `durable` mode and with every other provider the wait returns at once. When Mito drops a region it calls `release_region` after `obsolete`, as described under *Entry ids*; the default of that method does nothing, so the other providers handle a dropped region as before. Neither a release nor an `obsolete` of an object store region that fails, the latter for instance on a store that the release of another region poisoned, fails the drop, since nothing sends the drop of a removed region again and a caller that drops several regions, such as the metric engine with its data and metadata regions, would leave the others undone: the region's runtime state is cleaned up and its files are scheduled for deletion as for any drop, and the failure is logged; an `obsolete` that fails for another provider fails the drop as before. A store that is neither stopped nor poisoned keeps the record and writes it with its next object; a stopped or poisoned store keeps none, and the region stays unreleased after a restart. The store carries the record in a later object, as described under *Entry ids*; a crash before any later create succeeds loses it, and the dropped region's segments then hold the reclaim boundary after the restart. ## Datanode wiring and lifecycle @@ -215,9 +218,10 @@ on_corrupted_segment = "skip" ## Compatibility - `WalOptions::ObjectStore` is a new persisted variant. Existing `raft_engine`, `kafka` and `noop` options decode unchanged; a region's options never change provider implicitly. -- The object format is version 1 and frozen once the provider is released; a version bump after that is a new decoder, not an in-place change. The version 1 header layout was revised before release to carry the epoch, the link, the reclaim boundary and the header CRC. No object in an earlier layout exists, since the provider has not been enabled anywhere. +- The object format is version 1 and frozen once the provider is released; a version bump after that is a new decoder, not an in-place change. The version 1 header layout was revised before release to carry the epoch, the link, the reclaim boundary and the header CRC. No object in an earlier layout exists, since the provider has not been enabled anywhere. Release records add no format version: they are the segment of a reserved region in a version 1 object, so a build that predates them still opens the prefix. Such a build indexes that segment as a region it never opens, which holds its reclaim boundary at the first release until the release records are understood again. A frozen object with release records under `tests/compatibility/wal/object_store` is checked byte for byte by the format tests; the SQL compatibility runner owns `wal.provider` and does not cover this provider. - The configuration tag carries `experimental_`; the persisted tag does not, so promoting the provider to stable is a configuration rename with an alias, not a metadata migration. - `LogStore::wait_durable` has a default implementation that returns at once, so Raft Engine, Kafka and the noop provider keep their own acknowledgement and persistence semantics; the durability barrier binds only the object store provider. +- `LogStore::release_region` has a default implementation that does nothing, so Raft Engine, Kafka and the noop provider keep their handling of a dropped region. - The object-sequence-major id scheme needs no migration of existing objects. A prefix written under the contiguous scheme keeps its objects, and the sequence allocation rule under *Entry ids* keeps new ids above the old ones. Garbage collection reasons from footers and from the reclaim boundary, not from the id encoding, so it applies to objects of either scheme. - `on_corrupted_segment` is a new configuration option with a default, so existing configurations decode unchanged; it changes what a read does with a damaged segment, not what is written. @@ -238,12 +242,13 @@ 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. | 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. | +| A dropped region holds no object after a restart that never opens it | Unit tests of the release record an object carries and the boundary it allows; of recovery seeding the watermark from the release record before any boundary is computed, with collection proceeding once the remaining region is flushed; of a failed release carried by a timer tick on an idle prefix and by a `durable` stop, which reports the record lost when its create fails too; of the release of a region without segments, and of a region id used again after its release; of a release record that outlives a restart together with the segment it covers and goes once both are below the boundary, deleted out of order; the fault-matrix cases of a release, which return only once its object is durable and indexed, fail with its create, and fail at once while the store is failing; the Mito test that drops one of two regions, restarts gracefully, abruptly, or after a release that failed, with nothing written in between, then flushes the other region and writes once more; the metric engine test of a physical region whose data and metadata regions wrote to the WAL, dropped while the release of the data region fails, while both releases fail, and while the release of the data region meets a conflicting object that poisons the store, which still drops both regions and forgets the physical region, and, in the first two cases, after which a store opened later keeps no segment of either; the frozen object with release records. | Recovery seeds the release, and after the flush and the next write every object written before the restart is deleted. | +| 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. | 42 cases: 24 for the eleven rows of the matrix, five of them for the create of a release record, and 18 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, 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. +- 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` and releases 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, 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. @@ -322,7 +327,9 @@ Simpler catalog, but every region then produces its own timer-driven object, so - Write latency under the `durable` mode is the time until the batch seals (up to `flush_interval`, 100 ms by default), plus waiting for every earlier object, plus one PUT. A serial writer waits for the next seal, so its extra latency over Raft Engine is the interval minus the base insert cost modulo the interval, and concurrency does not close the gap: measured within one session on a laptop it was about 23% of throughput at concurrency 1 and about 28% at concurrency 8, and on S3 against Raft Engine on a local NVMe store the `durable` mode reached 31% of Raft Engine's throughput at concurrency 1 and 54% at concurrency 8, a serial writer being bound to one batch per interval, while the `enqueued` mode matched it, see *Write path measurements* and *Cloud write path measurements*. - Every object costs a request. Timer-triggered sealing creates at most one object per interval per node, ten per second at the default; the `enqueued` mode can seal earlier when the unpersisted backlog reaches a threshold, and the batch seals on size only when the admitted bytes reach `max_batch_bytes` before the next tick. Under sustained load on S3 the timer ceiling was reached, about 600 creates per minute per node, about $0.18 per node-hour at the S3 standard list price. -- Garbage collection keeps every object at or above the reclaim boundary, so a region that is written but rarely flushed, or never flushed, holds the boundary at its first object and keeps every later object the prefix shares with other regions; recovery time grows with the number of retained objects. A prefix that receives no appends cannot persist a higher boundary, so an idle prefix is collected only after its next object. +- Garbage collection keeps every object at or above the reclaim boundary, so a region that is written but rarely flushed, or never flushed, holds the boundary at its first object and keeps every later object the prefix shares with other regions; recovery time grows with the number of retained objects. A prefix that receives no appends cannot persist a higher boundary, so an idle prefix is collected only after its next object; a release seals that object itself. +- A dropped region stays unreleased, and holds the reclaim boundary for good, after a crash between the deletion of its table's metadata and a durable release record, or after a crash that follows a failed release before any later create succeeds; `indexed_objects` then only grows. The first window is a gap of the drop procedure for every WAL provider, issue #9577. See *Entry ids*. +- A release costs one object of its own and waits for every object sealed before it, in the `enqueued` mode as well, so dropping a region waits for the uploads ahead of its release record; when they fail, or the store is stopped or poisoned, the drop succeeds all the same and logs the failure; a stopped or poisoned store leaves the region unreleased. - Every open of the store writes one start object without segments, one extra PUT per open, and every object carries a 54-byte header. Orphans, the objects of failed batches that were stored anyway, hold storage together with the rows Mito wrote again after those failures until garbage collection deletes them. - Object-sequence-major entry ids consume 20 bits per object for positions, which caps a region at about a million entries per object; the batch seals when the cap is reached. Sequence numbers are 44 bits wide as a result, enough for one object per second for half a million years. @@ -344,12 +351,13 @@ Sealing may only exclude objects whose writes were never acknowledged, which tak **Corruption.** Segment-level skipping with the hole recorded per region is implemented, see *Object format*. *Proposed*: cluster mode routes the WAL hole event to the metasrv so that a follower or a takeover knows the region needs a fresh replica rather than a replay. -**Garbage collection across nodes.** Within one prefix, garbage collection is implemented as described under *Entry ids*: watermark from each region's manifest, per-segment footer comparison, a reclaim boundary persisted in every header. A source prefix that still holds segments of regions which have moved away cannot be trimmed by the source alone, because it no longer learns those regions' watermarks. Either the target reports to the metasrv, after each flush, the source sequence below which it no longer needs the region, and the metasrv drives deletion, or a cluster-level janitor collects the watermarks; which of the two is the last of the *Unresolved questions*. +**Garbage collection across nodes.** Within one prefix, garbage collection is implemented as described under *Entry ids*: watermark from each region's manifest, per-segment footer comparison, a reclaim boundary persisted in every header. A source prefix that still holds segments of regions which have moved away cannot be trimmed by the source alone, because it no longer learns those regions' watermarks. Either the target reports to the metasrv, after each flush, the source sequence below which it no longer needs the region, and the metasrv drives deletion, or a cluster-level janitor collects the watermarks; which of the two is the last of the *Unresolved questions*. Either way reclamation consults the metadata for the regions that still exist, as Kafka's pruning by the metasrv does, which also lets go of a dropped region whose release never became durable, see *Entry ids*. # Future work 1. Probe the next sequence with a GET from the last known sequence instead of listing the whole prefix at recovery; on a prefix with many objects the probe is cheaper than the listing. 2. Distributed mode, which needs metasrv-side allocation of node ids and generations, the takeover chain in `WalOptions`, read-only replay of a foreign prefix, and cross-node garbage collection, as sketched under *Alignment with cluster mode*. +3. Close the crash window of the drop procedure, issue #9577 (https://github.com/GreptimeTeam/greptimedb/issues/9577), which also leaves the region's WAL unreleased, see *Entry ids*. # Unresolved questions diff --git a/src/log-store/src/metrics.rs b/src/log-store/src/metrics.rs index c787820aaeb..8f923615964 100644 --- a/src/log-store/src/metrics.rs +++ b/src/log-store/src/metrics.rs @@ -111,6 +111,13 @@ lazy_static! { "object store logstore failed object deletes total", ) .unwrap(); + /// Counter of release records in the objects the object store logstore + /// indexed, including records carried over from batches that failed. + pub static ref METRIC_OBJECT_STORE_WAL_RELEASE_RECORDS_TOTAL: IntCounter = register_int_counter!( + "greptime_logstore_object_store_wal_release_records_total", + "object store logstore release records made durable total", + ) + .unwrap(); /// How long an object of the object store logstore takes to become durable /// once its batch is sealed, sampled for every conditional create that @@ -294,7 +301,8 @@ lazy_static! { /// 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. + /// fail with a retryable error. Releases of dropped regions failed in + /// that state count too; their records wait for the next object. 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", diff --git a/src/log-store/src/object_store_wal.rs b/src/log-store/src/object_store_wal.rs index 6a25afb3e77..68046487873 100644 --- a/src/log-store/src/object_store_wal.rs +++ b/src/log-store/src/object_store_wal.rs @@ -51,11 +51,19 @@ //! The reclaim boundary of an object states that every object with a lower //! sequence was fully flushed when the object was sealed, so the chain need //! not reach below it. Every present object below the boundary of the last -//! indexed object is deleted, on the chain or not. +//! indexed object is deleted, on the chain or not. A region that is dropped, +//! or whose entries are all made obsolete, is released: a release record of +//! the region goes into an object sealed at once, as the single entry of a +//! segment of a reserved region id, so the object keeps format version 1. +//! Recovery reads those segments of the chain and seeds the obsolete +//! watermarks from them, so a store opened later collects the objects of a +//! region that is never opened again. //! //! Recovery lists the objects, reads and verifies only the header, trailer and //! footer of each, and indexes the footers of the chain in sequence order to -//! rebuild the object catalog, which rejects a sequence it already holds. +//! rebuild the object catalog, which rejects a sequence it already holds. Of +//! the segments it reads only those of release records, and one that does not +//! decode fails recovery. //! Segments are read and checksummed only when a read decodes them; a segment //! that does not decode is fetched once more and then skipped as a hole of its //! region or fails the read, as `on_corrupted_segment` says. diff --git a/src/log-store/src/object_store_wal/catalog.rs b/src/log-store/src/object_store_wal/catalog.rs index 70f7aa5b329..61b4d2c09cc 100644 --- a/src/log-store/src/object_store_wal/catalog.rs +++ b/src/log-store/src/object_store_wal/catalog.rs @@ -25,7 +25,7 @@ use crate::error::{ CorruptedWalObjectSnafu, InvalidWalEntryRangeSnafu, Result, WalObjectSequenceExhaustedSnafu, }; use crate::object_store_wal::batch::{OBJECT_SEQ_LIMIT, sequence_floor}; -use crate::object_store_wal::format::FooterEntry; +use crate::object_store_wal::format::{FooterEntry, RELEASE_REGION_ID}; /// Indexes objects by sequence and, per region, the objects that hold entries /// of that region. @@ -165,7 +165,9 @@ impl ObjectCatalog { /// /// A segment is live when its maximum entry id is above the watermark of /// its region, or its region has no watermark; a region whose watermark - /// is [`EntryId::MAX`] holds no live segment. Watermarks never move down + /// is [`EntryId::MAX`] holds no live segment, and neither does + /// [`RELEASE_REGION_ID`], whose records recovery reads only from the + /// objects it indexes. Watermarks never move down /// and a region first appears in an object above the boundary, so the /// scan starts at the previous boundary and never reads an object twice /// once the boundary has passed it. @@ -176,9 +178,10 @@ impl ObjectCatalog { ) -> u64 { let live = self.objects.range(from..).find(|(_, footer)| { footer.iter().any(|entry| { - obsolete_entry_ids - .get(&entry.region_id) - .is_none_or(|obsolete| entry.max_entry_id > *obsolete) + entry.region_id != RELEASE_REGION_ID + && obsolete_entry_ids + .get(&entry.region_id) + .is_none_or(|obsolete| entry.max_entry_id > *obsolete) }) }); match live { @@ -511,7 +514,10 @@ mod tests { catalog .insert_object( 2, - vec![footer_entry(region_one, entry_id(2, 1), entry_id(2, 1))], + vec![ + footer_entry(region_one, entry_id(2, 1), entry_id(2, 1)), + footer_entry(RELEASE_REGION_ID, entry_id(2, 1), entry_id(2, 1)), + ], ) .unwrap(); catalog @@ -542,6 +548,7 @@ mod tests { &[(region_one, entry_id(1, 2)), (region_two, entry_id(1, 1))] ) ); + // The segment of release records never holds the boundary. assert_eq!( 3, boundary( diff --git a/src/log-store/src/object_store_wal/format.rs b/src/log-store/src/object_store_wal/format.rs index 7a255711784..cbd35287050 100644 --- a/src/log-store/src/object_store_wal/format.rs +++ b/src/log-store/src/object_store_wal/format.rs @@ -77,6 +77,20 @@ pub(crate) struct Record { pub(crate) payload: Bytes, } +/// The region whose segment holds the release records of an object, as a +/// single entry whose payload [`encode_releases`] writes. It is an ordinary +/// segment, so a reader that predates release records reads the object as +/// any other; no region is ever assigned this id. +pub(crate) const RELEASE_REGION_ID: RegionId = RegionId::from_u64(u64::MAX); + +/// A release record: no entry of the region with an id at or below +/// `entry_id` is needed any more, `u64::MAX` releasing the region in full. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Release { + pub(crate) region_id: RegionId, + pub(crate) entry_id: u64, +} + /// Footer entry describing the segment of one region. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct FooterEntry { @@ -470,6 +484,51 @@ pub(crate) fn decode_segment(bytes: &[u8], entry: &FooterEntry) -> Result) -> Bytes { + let mut output = BytesMut::with_capacity(4 + releases.len() * 16); + output.put_u32(releases.len() as u32); + for (region_id, entry_id) in releases { + output.put_u64(region_id.as_u64()); + output.put_u64(*entry_id); + } + output.freeze() +} + +/// Decodes the payload [`encode_releases`] writes. +pub(crate) fn decode_releases(payload: &[u8]) -> Result> { + let mut reader = Reader::new(payload); + let count = reader.u32("release records")? as usize; + ensure!( + count > 0 && reader.remaining_len() == count * 16, + CorruptedWalObjectSnafu { + reason: format!( + "release records declare {count} records in {} bytes", + reader.remaining_len() + ), + } + ); + let mut releases = Vec::::with_capacity(count); + for _ in 0..count { + let region_id = RegionId::from_u64(reader.u64("release records")?); + ensure!( + releases + .last() + .is_none_or(|last| last.region_id < region_id), + CorruptedWalObjectSnafu { + reason: format!("release records are not strictly ordered at region {region_id}"), + } + ); + releases.push(Release { + region_id, + entry_id: reader.u64("release records")?, + }); + } + Ok(releases) +} + /// Lower bound on the length of an object: header, a footer holding only its /// entry count and the trailer. pub(crate) const MIN_OBJECT_LEN: usize = HEADER_LEN + 4 + TRAILER_LEN; @@ -1138,6 +1197,86 @@ mod tests { } } + /// An object with an entry of region (1, 1) and the release records of + /// regions (2, 1) and (3, 0), frozen under `tests/compatibility`. + const FIXTURE_RELEASE: &[u8] = + include_bytes!("../../../../tests/compatibility/wal/object_store/release_record.wal"); + + fn release_fixture_records() -> Vec { + let releases = BTreeMap::from([ + (RegionId::new(2, 1), (2 << 20) | 1), + (RegionId::new(3, 0), u64::MAX), + ]); + [ + (RegionId::new(1, 1), Bytes::from_static(b"a1")), + (RELEASE_REGION_ID, encode_releases(&releases)), + ] + .into_iter() + .map(|(region_id, payload)| Record { + region_id, + entry_id: (3 << 20) | 1, + payload, + }) + .collect() + } + + #[test] + fn test_format_matches_release_record_fixture() { + let header = Header { + object_seq: 3, + epoch: 2, + prev: Some(ChainLink { + object_seq: 2, + epoch: 2, + }), + reclaim_boundary: 1, + }; + let encoded = encode_object(header.clone(), &release_fixture_records()).unwrap(); + assert_eq!(FIXTURE_RELEASE, encoded.bytes.as_ref()); + + // The object is a version 1 object whose last segment holds the + // release records, so a reader that predates them decodes it. + assert_eq!([0, 1], FIXTURE_RELEASE[8..10]); + let decoded = decode_object(FIXTURE_RELEASE).unwrap(); + assert_eq!(header, decoded.header); + assert_eq!(release_fixture_records(), decoded.records); + assert_eq!( + vec![ + Release { + region_id: RegionId::new(2, 1), + entry_id: (2 << 20) | 1, + }, + Release { + region_id: RegionId::new(3, 0), + entry_id: u64::MAX, + }, + ], + decode_releases(&decoded.records[1].payload).unwrap() + ); + } + + #[test] + fn test_format_rejects_invalid_release_records() { + let one = BTreeMap::from([(RegionId::new(1, 1), 1)]); + let encoded = encode_releases(&one); + for (payload, reason) in [ + (&[0, 0, 0, 0][..], "declare 0 records in 0 bytes"), + ( + &encoded[..encoded.len() - 1], + "declare 1 records in 15 bytes", + ), + ] { + assert_corrupted(decode_releases(payload), reason); + } + let mut unordered = encode_releases(&BTreeMap::from([ + (RegionId::new(1, 1), 1), + (RegionId::new(1, 2), 1), + ])) + .to_vec(); + unordered[4..12].copy_from_slice(&RegionId::new(1, 2).as_u64().to_be_bytes()); + assert_corrupted(decode_releases(&unordered), "not strictly ordered"); + } + fn footer_ids(entry: &FooterEntry) -> (u64, u64, u32) { (entry.min_entry_id, entry.max_entry_id, entry.entry_count) } diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index f233f73bdea..fc32bef707e 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -55,7 +55,7 @@ use crate::metrics::{ 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_RELEASE_RECORDS_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, }; @@ -63,8 +63,9 @@ use crate::object_store_wal::batch::{OBJECT_SEQ_LIMIT, OpenBatch, entry_id, sequ use crate::object_store_wal::catalog::ObjectCatalog; use crate::object_store_wal::format::{ ChainLink, EncodedObject, FixedTrailer, FooterEntry, HEADER_LEN, Header, MIN_OBJECT_LEN, - Record, TRAILER_LEN, decode_footer, decode_header, decode_segment, decode_trailer, - encode_object, footer_range, object_len, verify_segment_ranges, + RELEASE_REGION_ID, Record, Release, TRAILER_LEN, decode_footer, decode_header, decode_releases, + decode_segment, decode_trailer, encode_object, encode_releases, footer_range, object_len, + verify_segment_ranges, }; use crate::object_store_wal::io::{ListedObject, ObjectStoreIo, PutResult}; @@ -138,7 +139,9 @@ const RECOVERY_TAIL_WINDOW: usize = 64 * 1024; /// 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. +/// boundary of the last indexed object are deleted in the background. A +/// release of a region seals an object with a release record of the region at +/// once and returns, in both modes, once that object is durable and indexed. pub struct ObjectStoreLogStore { prefix: String, ack_mode: AckMode, @@ -315,6 +318,7 @@ impl ObjectStoreLogStore { max_epoch, reclaim_boundary, unindexed, + released_entry_ids, recovered_objects, } = recover(io.as_ref()).await?; // Every epoch is one above the sequence of the start object its @@ -349,10 +353,11 @@ impl ObjectStoreLogStore { })?; METRIC_OBJECT_STORE_WAL_INDEXED_OBJECTS.inc(); METRIC_OBJECT_STORE_WAL_INDEXED_BYTES.add(object_len(&[]) as i64); - // No watermark is recorded yet. - let writer_reclaim_boundary = catalog.reclaim_boundary(reclaim_boundary, &HashMap::new()); + // The release records are the only watermarks recorded yet. + let writer_reclaim_boundary = + catalog.reclaim_boundary(reclaim_boundary, &released_entry_ids); let catalog = Arc::new(RwLock::new(catalog)); - let obsolete_entry_ids = ObsoleteEntryIds::default(); + let obsolete_entry_ids = Arc::new(Mutex::new(released_entry_ids)); let wal_holes = WalHoles::default(); let deleting = Arc::new(DeletingObjects::default()); let terminal_error = TerminalError::default(); @@ -403,6 +408,7 @@ impl ObjectStoreLogStore { reclaim_boundary: writer_reclaim_boundary, durable_reclaim_boundary: reclaim_boundary, unindexed: unindexed.into_iter().chain(moved_past).collect(), + failed_releases: BTreeMap::new(), creating: HashSet::new(), deleting: deleting.clone(), deletes: FuturesUnordered::new(), @@ -521,6 +527,30 @@ impl ObjectStoreLogStore { None => Ok(()), } } + + /// Releases the region: its watermark moves up to the release in memory + /// and a release record of it goes into an object sealed at once, and the + /// call returns once that object is durable and indexed, in both + /// acknowledgement modes, or with the error that failed it. A `full` + /// release reaches [`EntryId::MAX`]; any other the largest id the store + /// handed out for the region, which every entry of the region it holds is + /// at or below. When the object fails, the next sealed batch carries the + /// record again. Recovery seeds the watermark from the record, so a store + /// opened later collects the objects of a region that is never opened + /// again. + async fn release(&self, region_id: RegionId, full: bool) -> Result<()> { + let (response_tx, response_rx) = oneshot::channel(); + self.command_tx + .send(Command::Release { + region_id, + full, + response: response_tx, + }) + .await + .ok() + .context(ObjectStoreWalStoppedSnafu)?; + response_rx.await.ok().context(ObjectStoreWalStoppedSnafu)? + } } /// The transient object store error of a create that a testing hook fails. @@ -607,6 +637,12 @@ impl ObjectStoreLogStore { self.creates_fail.store(true, Ordering::Release); } + /// Lets the creates that run from now on write again after + /// [`fail_creates`](Self::fail_creates). + pub fn allow_creates(&self) { + self.creates_fail.store(false, Ordering::Release); + } + /// Makes the next create that writes its object report a transient object /// store error afterwards, so the object exists although its create /// failed. @@ -660,8 +696,8 @@ impl ObjectStoreLogStore { /// Waits until no object delete is in flight. The deletes a collection /// starts are in flight by the time the `obsolete`, `obsolete_all`, - /// [`seal_open_batch`](Self::seal_open_batch) or `durable` append that - /// started it returns. + /// `release_region`, [`seal_open_batch`](Self::seal_open_batch) or + /// `durable` append that started it returns. pub async fn wait_for_garbage_collection(&self) { self.deleting.wait_until_empty().await; } @@ -888,6 +924,7 @@ impl LogStore for ObjectStoreLogStore { let regions = catalog .objects_in_order() .flat_map(|(_, footer)| footer.iter().map(|entry| entry.region_id)) + .filter(|region_id| *region_id != RELEASE_REGION_ID) .collect::>(); Ok(regions .into_iter() @@ -913,7 +950,9 @@ impl LogStore for ObjectStoreLogStore { /// ids under the next sequence the floor cannot move, and the call fails /// with [`Error::WalObjectSequenceUnsettled`] and records neither. The /// watermark is kept in memory; the caller re-establishes it after a - /// restart. In the `enqueued` mode an `entry_id` this store handed out + /// restart unless it released the region, see + /// [`release_region`](LogStore::release_region). In the `enqueued` mode + /// an `entry_id` this store handed out /// needs no floor: it is never handed out again while the store runs. async fn obsolete( &self, @@ -949,23 +988,19 @@ impl LogStore for ObjectStoreLogStore { } } - /// Hides every entry of the region in memory and lets collection delete - /// its objects. The caller re-establishes the watermark after a restart. + /// Hides every entry of the region and lets collection delete its + /// objects, by a release in full that is durable once this returns. async fn obsolete_all(&self, provider: &Provider, region_id: RegionId) -> Result<()> { self.check_region(provider, region_id)?; - record_obsolete(&self.obsolete_entry_ids, region_id, EntryId::MAX); - let (response_tx, response_rx) = oneshot::channel(); - if self - .command_tx - .send(Command::Collect { - response: response_tx, - }) - .await - .is_ok() - { - let _ = response_rx.await; - } - Ok(()) + self.release(region_id, true).await + } + + /// Releases the region through the largest id handed out for it, durably + /// once this returns. Should the region id be used again, its new entries + /// have greater ids and stay readable. + async fn release_region(&self, provider: &Provider, region_id: RegionId) -> Result<()> { + self.check_region(provider, region_id)?; + self.release(region_id, false).await } fn entry( @@ -1031,8 +1066,15 @@ enum Command { watermark: EntryId, response: oneshot::Sender>, }, - /// Answered once the collection the recorded watermarks allow started. - Collect { response: oneshot::Sender<()> }, + /// Answered once the object holding the release record of the region is + /// durable and indexed, or with the reason it is not. A `full` release + /// reaches [`EntryId::MAX`], any other the largest id handed out for the + /// region. + Release { + region_id: RegionId, + full: bool, + response: oneshot::Sender>, + }, Stop { response: oneshot::Sender>, }, @@ -1074,11 +1116,15 @@ struct SealedBatch { reclaim_boundary: u64, bytes: Bytes, footer: Vec, + releases: Vec, first_admitted_at: Instant, sealed_at: Instant, /// Number of creates that were attempted for the batch. attempts: u32, waiters: Vec, + /// Callers of a release whose record the batch carries, answered once the + /// object is durable and indexed. + release_waiters: Vec>>, #[cfg(any(test, feature = "testing"))] seal_waiters: Vec>>, state: CreateState, @@ -1093,6 +1139,9 @@ impl SealedBatch { for waiter in self.waiters { let _ = waiter.response.send(Err(error())); } + for waiter in self.release_waiters { + let _ = waiter.send(Err(error())); + } #[cfg(any(test, feature = "testing"))] for waiter in self.seal_waiters { let _ = waiter.send(Err(error())); @@ -1113,7 +1162,8 @@ type CreateOutcome = (u64, Result); /// 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. +/// behind the backlog. Releases fail with it too, and their records wait for +/// a create. The first create that succeeds ends the state. struct Failing { /// The last error a create met, which rejected appends carry. error: Arc, @@ -1159,9 +1209,9 @@ impl Failing { /// | --- | --- | --- | --- | /// | 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 | -/// | 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 | +/// | transient error, `enqueued` mode | unchanged | already acknowledged, except a release, which waits for the repeat | healthy: the create is repeated with the same bytes after [`CREATE_RETRY_DELAY`], which an identical retry accepts | +/// | error that requires intervention, `enqueued` mode | unchanged | already acknowledged, except a release, which fails while its record stays in the backlog; later appends and releases 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, except a release, which fails with the stopped error | 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 | @@ -1169,8 +1219,9 @@ impl Failing { /// A create still in flight when its batch failed runs to completion and its /// outcome is ignored: whatever it stored is off the chain. After `stop` began /// nothing is admitted and no create starts, except that the `enqueued` mode -/// uploads its backlog; creates in flight run to completion and acknowledge -/// if they succeed. +/// uploads its backlog and the `durable` mode creates one object with the +/// release records of failed batches; creates in flight run to completion and +/// acknowledge if they succeed. /// /// 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 @@ -1180,12 +1231,13 @@ impl Failing { /// /// The actor also collects garbage. Every batch carries the reclaim boundary /// current when it was sealed, see [`ObjectCatalog::reclaim_boundary`], which -/// is recomputed after every `obsolete` and every indexed object and never -/// moves down. The boundary of the last indexed object is durable: every -/// present object below it, indexed or not, is deleted in the background, up -/// to [`MAX_IN_FLIGHT_DELETES`] at a time, after an `obsolete` or an indexing -/// that moved it. An indexed object is unindexed once its delete succeeded; a -/// failed delete is attempted again at the next collection. Nothing is +/// is recomputed after every `obsolete`, every release and every indexed +/// object and never moves down. The boundary of the last indexed object is +/// durable: every present object below it, indexed or not, is deleted in the +/// background, up to [`MAX_IN_FLIGHT_DELETES`] at a time, after an +/// `obsolete`, a release or an indexing that moved it. An indexed object is +/// unindexed once its delete succeeded; a failed delete is attempted again at +/// the next collection. Nothing is /// collected once stop began or the store is poisoned, and deletes in flight /// complete before `stop` returns. struct Actor { @@ -1240,6 +1292,11 @@ struct Actor { /// Sequences of objects that may be present but are not indexed: the /// objects recovery listed off the chain and the batches that failed. unindexed: BTreeSet, + /// Release records of batches that failed, which the next sealed batch + /// carries again, so a release whose call failed still becomes durable + /// while the store runs. A timer tick seals them when no entry is + /// admitted. + failed_releases: BTreeMap, /// Sequences whose create is in flight. creating: HashSet, deleting: Arc, @@ -1297,10 +1354,8 @@ impl Actor { Some(Command::Obsolete { region_id, entry_id, watermark, response }) => { self.handle_obsolete(region_id, entry_id, watermark, response); } - Some(Command::Collect { response }) => { - self.update_reclaim_boundary(); - self.collect(); - let _ = response.send(()); + Some(Command::Release { region_id, full, response }) => { + self.handle_release(region_id, full, response); } Some(Command::Stop { response }) => { self.handle_stop(response); @@ -1474,7 +1529,22 @@ impl Actor { /// Seals the open batch as the object `next_object_seq` and starts its /// create when a slot is free. Returns whether a batch was sealed. fn flush_open_batch(&mut self) -> bool { - if self.open_batch.is_empty() { + self.seal(Vec::new()) + } + + /// Seals the open batch together with `releases` and the release records + /// of failed batches as the object `next_object_seq`, even when the open + /// batch holds no entry, and starts its create when a slot is free. + /// Returns whether a batch was sealed. + /// + /// A failing store keeps the records of failed batches coalesced instead + /// of sealing them alone behind its backlog on every tick, until `stop` + /// seals them for its last attempt of the backlog. + fn seal(&mut self, releases: Vec) -> bool { + if self.open_batch.is_empty() + && releases.is_empty() + && (self.failed_releases.is_empty() || (self.failing.is_some() && self.stop.is_empty())) + { return false; } if let Some(error) = terminal(&self.terminal_error) { @@ -1506,7 +1576,12 @@ impl Actor { ), reclaim_boundary: self.reclaim_boundary, }; - let encoded = match encode_batch(header, entries) { + let mut records = std::mem::take(&mut self.failed_releases); + for release in releases { + let entry_id = records.entry(release.region_id).or_insert(release.entry_id); + *entry_id = (*entry_id).max(release.entry_id); + } + let encoded = match encode_batch(header, entries, &records) { Ok(encoded) => encoded, Err(error) => { self.poison(error); @@ -1534,10 +1609,18 @@ impl Actor { reclaim_boundary: self.reclaim_boundary, bytes: encoded.bytes, footer: encoded.footer, + releases: records + .into_iter() + .map(|(region_id, entry_id)| Release { + region_id, + entry_id, + }) + .collect(), first_admitted_at, sealed_at: Instant::now(), attempts: 0, waiters: std::mem::take(&mut self.pending), + release_waiters: Vec::new(), #[cfg(any(test, feature = "testing"))] seal_waiters: Vec::new(), state: CreateState::Pending, @@ -1549,11 +1632,13 @@ 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. A failing store only schedules the + /// the backlog of the `enqueued` mode and, once the stop command was + /// handled, the release records [`handle_stop`](Self::handle_stop) + /// seals in the `durable` 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 { + if self.is_stopped() && self.ack_mode == AckMode::Durable && self.stop.is_empty() { return; } if !self.is_stopped() @@ -1760,6 +1845,13 @@ impl Actor { if let Some((_, response)) = stalled { let _ = response.send(Err(failing.reject())); } + // So would a release. Its record stays in the backlog and becomes + // durable with it. + for batch in &mut self.sealed { + for waiter in batch.release_waiters.drain(..) { + let _ = waiter.send(Err(failing.reject())); + } + } } /// Ends the failing state, if the store is in it, because of `reason`. @@ -1808,6 +1900,7 @@ impl Actor { }; METRIC_OBJECT_STORE_WAL_INDEXED_OBJECTS.inc(); METRIC_OBJECT_STORE_WAL_INDEXED_BYTES.add(batch.bytes.len() as i64); + METRIC_OBJECT_STORE_WAL_RELEASE_RECORDS_TOTAL.inc_by(batch.releases.len() as u64); self.last_indexed = ChainLink { object_seq: batch.object_seq, epoch: self.epoch, @@ -1824,6 +1917,9 @@ impl Actor { last_entry_ids: waiter.last_entry_ids, })); } + for waiter in batch.release_waiters { + let _ = waiter.send(Ok(())); + } #[cfg(any(test, feature = "testing"))] for waiter in batch.seal_waiters { let _ = waiter.send(Ok(())); @@ -1884,8 +1980,9 @@ impl Actor { shared(&error) } }; - for batch in self.sealed.drain(..) { + for mut batch in std::mem::take(&mut self.sealed) { self.unindexed.insert(batch.object_seq); + self.keep_releases(&mut batch); batch.fail(failure); } self.reset_open_batch(); @@ -1893,13 +1990,32 @@ impl Actor { // 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 { + // its caller has arrived yet or not. So it does release records once + // stop began, which no later create carries. + if self.ack_mode == AckMode::Enqueued + || (!self.stop.is_empty() && !self.failed_releases.is_empty()) + { self.stop_error.get_or_insert(error); } self.leave_failing("its backlog was dropped"); } + /// Keeps the release records of a batch that failed for the next sealed + /// batch. + fn keep_releases(&mut self, batch: &mut SealedBatch) { + for release in batch.releases.drain(..) { + self.keep_release(release); + } + } + + fn keep_release(&mut self, release: Release) { + let entry_id = self + .failed_releases + .entry(release.region_id) + .or_insert(release.entry_id); + *entry_id = (*entry_id).max(release.entry_id); + } + /// Records `error` as terminal and fails every waiter that is not /// acknowledged with it, or with the stopped error if the store was /// stopped meanwhile. Creates in flight run to completion, but their @@ -1910,6 +2026,8 @@ impl Actor { fn poison(&mut self, error: Error) -> Arc { let error = set_terminal(&self.terminal_error, error); let stopped = self.is_stopped(); + let releases_lost = !self.failed_releases.is_empty() + || self.sealed.iter().any(|batch| !batch.releases.is_empty()); let failure = || { if stopped { ObjectStoreWalStoppedSnafu.build() @@ -1922,7 +2040,9 @@ impl Actor { } self.reset_open_batch(); self.fail_unacknowledged(failure); - if self.ack_mode == AckMode::Enqueued { + // As in `roll_back`, release records lost once stop began are + // reported by `stop` in both modes. + if self.ack_mode == AckMode::Enqueued || (!self.stop.is_empty() && releases_lost) { self.stop_error.get_or_insert(error.clone()); } self.leave_failing("the store is poisoned"); @@ -2014,6 +2134,63 @@ impl Actor { let _ = response.send(result); } + /// Moves the watermark of the region up to the release and seals the + /// open batch with a release record of the region; `response` is + /// answered once that object is durable and indexed. The batch carries + /// the reclaim boundary the watermark allows, so the object makes both + /// durable at once. + /// + /// The watermark is not clamped to the durable entry id in the `enqueued` + /// mode: a released region is never replayed, and the record is durable + /// only once every object sealed before it is. A release arrives as a + /// command, so [`MAX_SEALED_BATCHES`] does not hold it back; every caller + /// waits for its own object, so releases add at most one batch per caller. + fn handle_release( + &mut self, + region_id: RegionId, + full: bool, + response: oneshot::Sender>, + ) { + if self.is_stopped() { + let _ = response.send(Err(ObjectStoreWalStoppedSnafu.build())); + return; + } + if let Some(error) = terminal(&self.terminal_error) { + let _ = response.send(Err(shared(&error))); + return; + } + let entry_id = if full { + EntryId::MAX + } else { + self.issued_entry_ids.get(®ion_id).copied().unwrap_or(0) + }; + record_obsolete(&self.obsolete_entry_ids, region_id, entry_id); + self.update_reclaim_boundary(); + self.collect(); + let release = Release { + region_id, + entry_id, + }; + // A failing store admits nothing behind its backlog: the release + // fails at once, and its record is kept like that of a batch that + // failed, sealed once a create succeeds. + if let Some(failing) = &mut self.failing { + let _ = response.send(Err(failing.reject())); + self.keep_release(release); + return; + } + if self.seal(vec![release]) + && let Some(batch) = self.sealed.back_mut() + { + batch.release_waiters.push(response); + return; + } + // Sealing failed and poisoned the store. + if let Some(error) = terminal(&self.terminal_error) { + let _ = response.send(Err(shared(&error))); + } + } + /// Raises the reclaim boundary to what the catalog and the obsolete /// watermarks allow. fn update_reclaim_boundary(&mut self) { @@ -2162,7 +2339,8 @@ impl Actor { /// 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 + /// batches whose create has not started, and seals the release records of + /// failed and dropped batches once more; 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>) { @@ -2183,10 +2361,12 @@ impl Actor { .iter() .position(|batch| matches!(batch.state, CreateState::Pending)) { - for batch in self.sealed.drain(index..) { + for mut batch in self.sealed.split_off(index) { + self.keep_releases(&mut batch); batch.fail(|| ObjectStoreWalStoppedSnafu.build()); } } + self.seal(Vec::new()); } AckMode::Enqueued => { self.flush_open_batch(); @@ -2486,7 +2666,19 @@ fn record_skipped_segment( .or_insert(hole); } -fn encode_batch(header: Header, entries: Vec) -> Result { +/// Encodes the entries of a batch, and its release records as the single +/// entry of [`RELEASE_REGION_ID`], whose id is at position 1 of the object; +/// segments are ordered by region id, so it is the last segment. +fn encode_batch( + header: Header, + entries: Vec, + releases: &BTreeMap, +) -> Result { + let release_record = (!releases.is_empty()).then(|| Record { + region_id: RELEASE_REGION_ID, + entry_id: entry_id(header.object_seq, 1), + payload: encode_releases(releases), + }); let records = entries .into_iter() .map(|entry| Record { @@ -2494,6 +2686,7 @@ fn encode_batch(header: Header, entries: Vec) -> Result { entry_id: entry.entry_id(), payload: Bytes::from(entry.into_bytes()), }) + .chain(release_record) .collect::>(); encode_object(header, &records) } @@ -2513,6 +2706,9 @@ struct Recovered { reclaim_boundary: u64, /// Present objects that are not indexed, in sequence order. unindexed: Vec, + /// The entry id each region is released through, from the release + /// records of the indexed objects, see [`read_releases`]. + released_entry_ids: HashMap, /// Number of objects whose header and footer were read. recovered_objects: u64, } @@ -2535,14 +2731,71 @@ struct FetchedObject { /// Rebuilds the catalog from object headers and footers, /// so recovery costs a few small reads per object however large the objects -/// are. Segments are not read; a segment checksum is verified by the read that -/// decodes it. Footers are fetched for up to [`RECOVERY_CONCURRENCY`] objects +/// are. Segments are not read, except those of release records, see +/// [`read_releases`]; a segment checksum is verified by the read that decodes +/// it. Footers are fetched for up to [`RECOVERY_CONCURRENCY`] objects /// at a time and indexed in sequence order, so the catalog checks the entry /// ranges of every object against its predecessors like a sequential replay. /// Only the objects on the chain [`select_chain`] picks are indexed. async fn recover(io: &dyn WalObjectIo) -> Result { let objects = io.list().await?; - finish_recovery(fetch_footers(io, objects, RECOVERY_CONCURRENCY).await?) + let mut recovered = finish_recovery(fetch_footers(io, objects, RECOVERY_CONCURRENCY).await?)?; + recovered.released_entry_ids = read_releases(io, &recovered.catalog).await?; + Ok(recovered) +} + +/// Reads the release records of the indexed objects, which seed the +/// watermarks of their regions: one range read of the segment of +/// [`RELEASE_REGION_ID`] per object that holds one. A segment that cannot be +/// fetched or does not decode fails recovery, whatever `on_corrupted_segment` +/// says: skipping it would lose the watermarks it seeds. +/// +/// A release record needs no bookkeeping of its own. It is sealed once every +/// entry it covers was admitted, so it lies in the object holding the last +/// segment it covers or in a later one. Collection deletes, in any order, +/// only objects below the durable reclaim boundary, and recovery indexes only +/// objects at or above the boundary of the tip, which a start object carries +/// forward. So recovery indexes a release record whenever it indexes a +/// segment the record covers, and once the record is below the boundary so is +/// every segment it covers, which recovery never indexes again, whether +/// collection deleted it yet or not. A release in full also covers entries a +/// region would be assigned afterwards, so it is made only for a region that +/// writes nothing more. +async fn read_releases( + io: &dyn WalObjectIo, + catalog: &ObjectCatalog, +) -> Result> { + let segments = catalog + .objects_in_order() + .filter_map(|(object_seq, footer)| { + footer + .iter() + .find(|entry| entry.region_id == RELEASE_REGION_ID) + .map(|entry| (object_seq, entry.clone())) + }) + .collect::>(); + let releases = futures::stream::iter(segments) + .map(|(object_seq, entry)| async move { + let bytes = io + .get_range(object_seq, entry.segment_offset, entry.segment_len) + .await?; + decode_segment(&bytes, &entry) + .and_then(|records| decode_releases(&records[0].payload)) + .with_context(|_| InvalidWalObjectSnafu { + path: io.object_path(object_seq), + }) + }) + .buffered(RECOVERY_CONCURRENCY) + .try_collect::>() + .await?; + let mut released_entry_ids = HashMap::new(); + for release in releases.into_iter().flatten() { + let entry_id = released_entry_ids + .entry(release.region_id) + .or_insert(release.entry_id); + *entry_id = (*entry_id).max(release.entry_id); + } + Ok(released_entry_ids) } /// Indexes the chain of the fetched objects. Objects off the chain, orphans @@ -2627,6 +2880,7 @@ fn finish_recovery(fetched: FetchedObjects) -> Result { max_epoch, reclaim_boundary, unindexed, + released_entry_ids: HashMap::new(), recovered_objects, }) } @@ -3417,6 +3671,7 @@ mod tests { assert_eq!(expected.max_epoch, actual.max_epoch); assert_eq!(expected.reclaim_boundary, actual.reclaim_boundary); assert_eq!(expected.unindexed, actual.unindexed); + assert_eq!(expected.released_entry_ids, actual.released_entry_ids); } fn catalog_contents(catalog: &ObjectCatalog) -> Vec<(u64, Vec)> { @@ -3600,8 +3855,8 @@ mod tests { let last = bytes.len() - 1; bytes[last] ^= 1; }), - ("unsupported format version 2", |bytes| { - bytes[8..10].copy_from_slice(&2u16.to_be_bytes()); + ("unsupported format version 3", |bytes| { + bytes[8..10].copy_from_slice(&3u16.to_be_bytes()); }), ("footer checksum mismatch", |bytes| { let (trailer, _) = footer_of(bytes); @@ -4320,19 +4575,10 @@ mod tests { assert_eq!(20, latest(&store, region(1))); store.obsolete(&p, region(1), 10).await.unwrap(); assert_eq!(remaining, read_entries(&store, region(1), 0).await); - store.obsolete_all(&p, region(1)).await.unwrap(); - store.obsolete(&p, region(1), 0).await.unwrap(); - assert_eq!( - Vec::::new(), - read_entries(&store, region(1), 0).await - ); - assert_eq!(20, latest(&store, region(1))); - assert_eq!( - expected_entries(region(2), &[(10, "e10")]), - read_entries(&store, region(2), 0).await - ); store.stop().await.unwrap(); - let reopened = open(object_store, &eager()).await; + + // A watermark is kept in memory only. + let reopened = open(object_store.clone(), &eager()).await; assert_eq!(all, read_entries(&reopened, region(1), 0).await); reopened.obsolete(&p, region(1), 20).await.unwrap(); assert_eq!( @@ -4340,6 +4586,31 @@ mod tests { read_entries(&reopened, region(1), 0).await ); assert_eq!(20, latest(&reopened, region(1))); + + // A release in full survives the restart, and a later watermark does + // not move it down. + reopened.obsolete_all(&p, region(1)).await.unwrap(); + reopened.obsolete(&p, region(1), 0).await.unwrap(); + assert_eq!( + Vec::::new(), + read_entries(&reopened, region(1), 0).await + ); + assert_eq!(20, latest(&reopened, region(1))); + reopened.stop().await.unwrap(); + let reopened = open(object_store, &eager()).await; + assert_eq!( + Vec::::new(), + read_entries(&reopened, region(1), 0).await + ); + reopened.obsolete(&p, region(1), 0).await.unwrap(); + assert_eq!( + Vec::::new(), + read_entries(&reopened, region(1), 0).await + ); + assert_eq!( + expected_entries(region(2), &[(10, "e10")]), + read_entries(&reopened, region(2), 0).await + ); reopened.stop().await.unwrap(); } @@ -7477,22 +7748,25 @@ mod tests { assert_eq!(2, reclaim_boundary_of(io.as_ref(), 3).await); assert_eq!(vec![2, 3], collected_object_seqs(&store).await); - // Region one dropped and region two flushed through object 2: object - // 4 carries the sequence after the last indexed object. + // Region two flushed through object 2 and region one released: the + // object of the release record, 4, carries the sequence after the + // last indexed object, and so does the next one. store.obsolete(&provider(two), two, id(2, 1)).await.unwrap(); store.obsolete_all(&provider(one), one).await.unwrap(); - append(&store, two, "b2").await.unwrap(); assert_eq!(4, reclaim_boundary_of(io.as_ref(), 4).await); assert_eq!(vec![4], collected_object_seqs(&store).await); + append(&store, two, "b2").await.unwrap(); + assert_eq!(5, reclaim_boundary_of(io.as_ref(), 5).await); + assert_eq!(vec![5], collected_object_seqs(&store).await); // A region that appears later does not move the boundary back. append(&store, three, "c1").await.unwrap(); - assert_eq!(4, reclaim_boundary_of(io.as_ref(), 5).await); - assert_eq!(vec![4, 5], collected_object_seqs(&store).await); + assert_eq!(5, reclaim_boundary_of(io.as_ref(), 6).await); + assert_eq!(vec![5, 6], collected_object_seqs(&store).await); // Other stores of the process count their deletes as well. - assert!(METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL.get() >= deleted + 4); + assert!(METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL.get() >= deleted + 5); assert_eq!( - expected_entries(two, &[(id(4, 1), "b2")]), + expected_entries(two, &[(id(5, 1), "b2")]), read_entries(&store, two, 0).await ); store.stop().await.unwrap(); @@ -7608,15 +7882,17 @@ mod tests { put_foreign(&object_store, 7, 100).await; append(&store, region_id, "a7").await.unwrap_err(); } else { - // The first poll queues the stop command; the actor answers - // the command of `obsolete_all` queued after it only once it + // The first poll queues the stop command; the actor refuses + // the release of `obsolete_all` queued after it only once it // handled the stop. let mut stopping = Box::pin(store.stop()); assert!((&mut stopping).now_or_never().is_none()); - store - .obsolete_all(&provider(region_id), region_id) - .await - .unwrap(); + assert_stopped( + &store + .obsolete_all(&provider(region_id), region_id) + .await + .unwrap_err(), + ); assert!((&mut stopping).now_or_never().is_none()); stop = Some(stopping); } @@ -7967,6 +8243,258 @@ mod tests { store.stop().await.unwrap(); } + fn spawn_release( + store: &Arc, + region_id: RegionId, + ) -> tokio::task::JoinHandle> { + let store = store.clone(); + tokio::spawn(async move { store.release_region(&provider(region_id), region_id).await }) + } + + fn release(region_id: RegionId, entry_id: EntryId) -> Release { + Release { + region_id, + entry_id, + } + } + + /// The release records of the object `object_seq`. + async fn releases_of(io: &dyn WalObjectIo, object_seq: u64) -> Vec { + let decoded = decode_object(&io.get(object_seq).await.unwrap()).unwrap(); + let record = decoded + .records + .iter() + .find(|record| record.region_id == RELEASE_REGION_ID) + .unwrap(); + assert_eq!(entry_id(object_seq, 1), record.entry_id); + decode_releases(&record.payload).unwrap() + } + + fn watermark(store: &ObjectStoreLogStore, region_id: RegionId) -> Option { + store + .obsolete_entry_ids + .lock() + .unwrap() + .get(®ion_id) + .copied() + } + + #[tokio::test] + async fn test_store_release_writes_a_record_of_the_largest_issued_id() { + let _serialized = SERIALIZED.lock().await; + let store = open(memory_store(), &eager()).await; + let (one, two) = (region(1), region(2)); + store + .append_batch(vec![entry(&store, one, "a1"), entry(&store, two, "b1")]) + .await + .unwrap(); + + // The release seals object 2 without any entry; its record releases + // region one through the largest id handed out for it, and it carries + // the boundary region two holds at object 1. + store.release_region(&provider(one), one).await.unwrap(); + assert_eq!( + vec![release(one, id(1, 1))], + releases_of(store.io.as_ref(), 2).await + ); + assert_eq!(1, reclaim_boundary_of(store.io.as_ref(), 2).await); + assert_eq!(vec![1, 2], collected_object_seqs(&store).await); + assert!(read_entries(&store, one, 0).await.is_empty()); + // The record is the segment of a reserved region, which names no + // namespace. + assert_eq!( + vec![provider(one), provider(two)], + store.list_namespaces().await.unwrap() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_recovery_seeds_watermarks_from_release_records() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + let (one, two) = (region(1), region(2)); + let store = open(object_store.clone(), &eager()).await; + // Object 1 holds entries of both regions and object 2 of region one. + // The release of region one is the last object written, and region + // two holds the boundary at object 1. + store + .append_batch(vec![entry(&store, one, "a1"), entry(&store, two, "b1")]) + .await + .unwrap(); + append(&store, one, "a2").await.unwrap(); + store.release_region(&provider(one), one).await.unwrap(); + assert_eq!(1, reclaim_boundary_of(store.io.as_ref(), 3).await); + store.stop().await.unwrap(); + + // Recovery seeds the watermark of region one, which nothing opens + // again, and a later watermark never moves it down. + let store = open(object_store, &eager()).await; + assert_eq!(vec![1, 2, 3, 4], collected_object_seqs(&store).await); + assert_eq!(Some(id(2, 1)), watermark(&store, one)); + assert!(read_entries(&store, one, 0).await.is_empty()); + store.obsolete(&provider(one), one, id(1, 1)).await.unwrap(); + assert_eq!(Some(id(2, 1)), watermark(&store, one)); + + // Once region two is flushed through object 1, no object holds a + // live segment: object 5 carries the boundary past all of them. + store.obsolete(&provider(two), two, id(1, 1)).await.unwrap(); + append(&store, two, "b2").await.unwrap(); + assert_eq!(5, reclaim_boundary_of(store.io.as_ref(), 5).await); + assert_eq!(vec![5], collected_object_seqs(&store).await); + assert_eq!( + expected_entries(two, &[(id(5, 1), "b2")]), + read_entries(&store, two, 0).await + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_carries_the_record_of_a_failed_release() { + let _serialized = SERIALIZED.lock().await; + let region_id = region(1); + + // The release fails with its create, and a tick seals the record + // again although nothing is appended. + let (io, _) = RecordingIo::over(memory_store()); + let store = open_over(io.clone(), &config(MIN_FLUSH_INTERVAL, u64::MAX)).await; + io.fail_next_put_persistently(); + store + .release_region(&provider(region_id), region_id) + .await + .unwrap_err(); + wait_until(|| store.catalog.read().unwrap().contains_object(2)).await; + assert_eq!( + vec![release(region_id, 0)], + releases_of(io.as_ref(), 2).await + ); + store.stop().await.unwrap(); + + // A `durable` stop creates the record once more, and reports it lost + // when that create fails or conflicts too. + for last_create in ["succeeds", "fails", "conflicts"] { + let object_store = memory_store(); + let store = open(object_store.clone(), &eager()).await; + append(&store, region_id, "a1").await.unwrap(); + store.fail_creates(); + store + .release_region(&provider(region_id), region_id) + .await + .unwrap_err(); + match last_create { + "succeeds" => store.allow_creates(), + "conflicts" => { + store.allow_creates(); + put_foreign(&object_store, 3, 2).await; + } + _ => {} + } + let stopped = store.stop().await; + let store = open(object_store, &eager()).await; + if last_create == "succeeds" { + stopped.unwrap(); + assert_eq!(Some(id(1, 1)), watermark(&store, region_id)); + } else { + let error = stopped.unwrap_err(); + let conflict = matches!(unwrap_shared(&error), Error::WalObjectConflict { .. }); + assert_eq!(last_create == "conflicts", conflict, "{error:?}"); + assert_eq!(None, watermark(&store, region_id)); + } + store.stop().await.unwrap(); + } + } + + #[tokio::test] + async fn test_store_releases_a_region_without_segments() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + let store = open(object_store.clone(), &eager()).await; + let io = store.io.clone(); + let (one, two, three) = (region(1), region(2), region(3)); + append(&store, two, "b1").await.unwrap(); + + // A region the store holds no entry of is released through zero and + // a release in full reaches the largest id, each in an object of its + // own; neither moves the boundary region two holds. + store.release_region(&provider(one), one).await.unwrap(); + store.obsolete_all(&provider(three), three).await.unwrap(); + assert_eq!(vec![release(one, 0)], releases_of(io.as_ref(), 2).await); + assert_eq!( + vec![release(three, EntryId::MAX)], + releases_of(io.as_ref(), 3).await + ); + assert_eq!(1, reclaim_boundary_of(io.as_ref(), 3).await); + assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await); + + // Region one written again stays readable across a restart. + append(&store, one, "a1").await.unwrap(); + store.stop().await.unwrap(); + let store = open(object_store, &eager()).await; + assert_eq!(Some(0), watermark(&store, one)); + assert_eq!(Some(EntryId::MAX), watermark(&store, three)); + assert_eq!( + expected_entries(one, &[(id(4, 1), "a1")]), + read_entries(&store, one, 0).await + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_release_record_lives_as_long_as_the_segments_it_covers() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + let (one, two) = (region(1), region(2)); + let store = open(object_store.clone(), &eager()).await; + // Object 1 holds entries of both regions, object 2 the release of + // region one and object 3 an entry of region two, which holds the + // boundary at object 1. + store + .append_batch(vec![entry(&store, one, "a1"), entry(&store, two, "b1")]) + .await + .unwrap(); + store.release_region(&provider(one), one).await.unwrap(); + append(&store, two, "b2").await.unwrap(); + assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await); + store.stop().await.unwrap(); + + // The record outlives a restart together with the segment it covers. + let (io, mut parked) = ParkedIo::parking_creates(object_store.clone()); + let store = open_over(io.clone(), &eager()).await; + assert_eq!(Some(id(1, 1)), watermark(&store, one)); + assert!(read_entries(&store, one, 0).await.is_empty()); + + // Region two flushed through object 1 lets object 5 carry boundary + // 3, below which the record and the segment go in any order: the + // record is deleted, the delete of the segment fails. + io.deletes_parked.store(true, Ordering::SeqCst); + store.obsolete(&provider(two), two, id(1, 1)).await.unwrap(); + append(&store, two, "b3").await.unwrap(); + assert_eq!(3, reclaim_boundary_of(io.as_ref(), 5).await); + let mut deletes = parked_operations(&mut parked, 2).await; + deletes.remove(&2).unwrap().send(true).unwrap(); + deletes.remove(&1).unwrap().send(false).unwrap(); + timeout(WAIT, store.wait_for_garbage_collection()) + .await + .unwrap(); + assert_eq!(vec![1, 3, 4, 5], object_seqs(io.as_ref()).await); + store.stop().await.unwrap(); + + // Recovery indexes nothing below boundary 3, which the start object + // carries forward, so the segment the record covered is never live + // again, and the first collection deletes it. + let plain = ObjectStoreIo::new(object_store.clone(), PREFIX).unwrap(); + let recovered = recover(&plain).await.unwrap(); + assert_eq!(3, recovered.reclaim_boundary); + assert_eq!(vec![1], recovered.unindexed); + assert!(recovered.released_entry_ids.is_empty()); + let store = open(object_store, &eager()).await; + assert_eq!(3, reclaim_boundary_of(&plain, 6).await); + assert_eq!(None, watermark(&store, one)); + store.obsolete(&provider(two), two, id(3, 1)).await.unwrap(); + assert_eq!(vec![3, 4, 5, 6], collected_object_seqs(&store).await); + store.stop().await.unwrap(); + } + /// Object access whose next create stores the object but reports a /// transient failure, as a create whose response is lost. struct LostResponseIo { @@ -8430,6 +8958,7 @@ mod tests { reads: METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count(), deleted_objects: METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL.get(), failed_deletes: METRIC_OBJECT_STORE_WAL_FAILED_DELETES_TOTAL.get(), + release_records: METRIC_OBJECT_STORE_WAL_RELEASE_RECORDS_TOTAL.get(), /// No read of the matrix meets a corrupted segment. skipped_segments: METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(), stalled_appends: METRIC_OBJECT_STORE_WAL_STALLED_APPENDS_TOTAL.get(), @@ -8488,6 +9017,15 @@ mod tests { object_bytes_of(&[(region(1), "a1")]) } + /// The length of an object that holds one release record alone. + fn release_object() -> u64 { + let releases = BTreeMap::from([(region(1), 0)]); + encode_batch(chain_header(1, 1, None), Vec::new(), &releases) + .unwrap() + .bytes + .len() as u64 + } + /// The sum and the count of a timing histogram before a forced wait. fn timer_before(histogram: &Histogram) -> (f64, u64) { (histogram.get_sample_sum(), histogram.get_sample_count()) @@ -8680,6 +9218,36 @@ mod tests { row: Some(Row::LastSequence), run: || Box::pin(create_takes_the_last_object_sequence()), }, + Case { + fault: "a release whose create succeeds", + row: Some(Row::Created), + run: || Box::pin(release_create_succeeds()), + }, + Case { + fault: "a release whose create fails transiently, `durable` mode", + row: Some(Row::TransientDurable), + run: || Box::pin(release_create_fails_transiently_durable()), + }, + Case { + fault: "a release whose create fails transiently, `enqueued` mode", + row: Some(Row::TransientEnqueued), + run: || Box::pin(release_create_fails_transiently_enqueued()), + }, + Case { + fault: "a release whose create the object store refuses with an error that requires intervention, `enqueued` mode", + row: Some(Row::InterventionEnqueued), + run: || Box::pin(release_create_refused_enqueued()), + }, + Case { + fault: "a release while the object store refuses the backlog, `enqueued` mode", + row: None, + run: || Box::pin(release_while_the_backlog_is_refused()), + }, + Case { + fault: "a release whose create finds an object of the same or a later epoch", + row: Some(Row::Conflict), + run: || Box::pin(release_create_finds_another_writer()), + }, Case { fault: "a start object that meets an object of an earlier epoch", row: Some(Row::StartMovesPast), @@ -8877,6 +9445,310 @@ mod tests { store.stop().await.unwrap(); } + /// A release seals an object of its own and returns once it is + /// indexed, in both modes. + async fn release_create_succeeds() { + for config in [eager(), enqueued(eager())] { + let (store, io, mut parked) = open_parking_creates(memory_store(), &config).await; + let before = Counters::sample(); + let region_id = region(1); + let released = spawn_release(&store, region_id); + let (object_seq, create) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + assert!(!released.is_finished()); + assert_indexed(io.as_ref(), &[0]).await; + create.send(true).unwrap(); + timeout(WAIT, released).await.unwrap().unwrap().unwrap(); + // Object 1 carries the boundary past the start object. + assert_eq!(vec![1], collected_object_seqs(&store).await); + assert_eq!( + vec![release(region_id, 0)], + releases_of(io.as_ref(), 1).await + ); + assert!(read_entries(&store, region_id, 1).await.is_empty()); + assert_indexed(io.as_ref(), &[1]).await; + assert_eq!( + Counters { + created_objects: 1, + seal_to_durable: 1, + sealed_objects: 1, + object_bytes: release_object(), + reads: 1, + deleted_objects: 1, + release_records: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + } + + /// The release fails with the create, its sequence is not reused, and + /// the release of the region again lands in the next object. + async fn release_create_fails_transiently_durable() { + let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await; + let before = Counters::sample(); + let region_id = region(1); + let released = spawn_release(&store, region_id); + let (object_seq, create) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + create.send(false).unwrap(); + let error = timeout(WAIT, released).await.unwrap().unwrap().unwrap_err(); + assert!( + matches!(unwrap_shared(&error), Error::WalObjectStore { .. }), + "unexpected error: {error:?}" + ); + assert!(terminal(&store.terminal_error).is_none()); + assert_eq!(vec![0], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + create_failures: 1, + sealed_objects: 1, + object_bytes: release_object(), + ..Counters::default() + }, + Counters::since(before) + ); + + let released = spawn_release(&store, region_id); + let (object_seq, create) = next_parked_operation(&mut parked).await; + assert_eq!(2, object_seq); + create.send(true).unwrap(); + timeout(WAIT, released).await.unwrap().unwrap().unwrap(); + assert_eq!(vec![2], collected_object_seqs(&store).await); + assert_indexed(io.as_ref(), &[2]).await; + assert_eq!( + Counters { + created_objects: 1, + seal_to_durable: 1, + create_failures: 1, + sealed_objects: 2, + object_bytes: 2 * release_object(), + deleted_objects: 1, + release_records: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// The store repeats the create, and the release waits for the repeat. + async fn release_create_fails_transiently_enqueued() { + let (store, io, mut parked) = + open_parking_creates(memory_store(), &enqueued(eager())).await; + let before = Counters::sample(); + let region_id = region(1); + let released = spawn_release(&store, region_id); + let (object_seq, create) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + create.send(false).unwrap(); + let (object_seq, repeat) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + assert!(!released.is_finished()); + 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: 1, + sealed_objects: 1, + object_bytes: release_object(), + ..Counters::default() + }, + Counters::since(before) + ); + + repeat.send(true).unwrap(); + timeout(WAIT, released).await.unwrap().unwrap().unwrap(); + assert_eq!(vec![1], collected_object_seqs(&store).await); + assert_indexed(io.as_ref(), &[1]).await; + assert_eq!( + Counters { + created_objects: 1, + seal_to_durable: 1, + create_failures: 1, + sealed_objects: 1, + object_bytes: release_object(), + deleted_objects: 1, + release_records: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// The object store denies access to the object of a release: the + /// store becomes failing and the release fails at once with the + /// retryable error; its record stays in the backlog and becomes + /// durable with the next create that succeeds. + async fn release_create_refused_enqueued() { + let (store, io, mut parked) = + open_parking_creates(memory_store(), &enqueued(eager())).await; + let before = Counters::sample(); + let region_id = region(1); + let released = spawn_release(&store, region_id); + let (object_seq, create) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + io.creates_refused.store(true, Ordering::SeqCst); + create.send(false).unwrap(); + let error = timeout(WAIT, released).await.unwrap().unwrap().unwrap_err(); + assert_rejected(&error, ErrorKind::PermissionDenied); + assert_eq!(1, failing_stores()); + assert!(terminal(&store.terminal_error).is_none()); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + create_failures: 1, + sealed_objects: 1, + object_bytes: release_object(), + rejected_appends: 1, + ..Counters::default() + }, + Counters::since(before) + ); + + let (object_seq, create) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + create.send(true).unwrap(); + wait_until(|| failing_stores() == 0).await; + assert_eq!(vec![1], collected_object_seqs(&store).await); + assert_eq!( + vec![release(region_id, 0)], + releases_of(io.as_ref(), 1).await + ); + assert_indexed(io.as_ref(), &[1]).await; + assert_eq!( + Counters { + created_objects: 1, + seal_to_durable: 1, + create_failures: 1, + sealed_objects: 1, + object_bytes: release_object(), + deleted_objects: 1, + release_records: 1, + rejected_appends: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// Releases that arrive while the store is failing fail at once with + /// the retryable error, and their record is kept coalesced: the + /// ticks that pass seal nothing behind the backlog. A stop seals the + /// record with the backlog, and recovery reads it back. + async fn release_while_the_backlog_is_refused() { + let object_store = memory_store(); + let (store, io, mut parked) = open_parking_creates( + object_store.clone(), + &enqueued(config(MIN_FLUSH_INTERVAL, 1)), + ) + .await; + let before = Counters::sample(); + let region_id = region(1); + append(&store, region_id, "a1").await.unwrap(); + io.creates_refused.store(true, Ordering::SeqCst); + next_parked_operation(&mut parked) + .await + .1 + .send(false) + .unwrap(); + wait_until(|| failing_stores() == 1).await; + for _ in 0..3 { + let error = store + .release_region(&provider(region_id), region_id) + .await + .unwrap_err(); + assert_rejected(&error, ErrorKind::PermissionDenied); + } + // The next attempt of the backlog comes after several ticks. + let (object_seq, probe) = next_parked_operation(&mut parked).await; + assert_eq!(1, object_seq); + assert!(parked.try_recv().is_err()); + probe.send(false).unwrap(); + wait_until(|| Counters::since(before).create_failures == 2).await; + assert_eq!(1, Counters::since(before).sealed_objects); + + let stop = { + let store = store.clone(); + tokio::spawn(async move { store.stop().await }) + }; + for create in parked_operations(&mut parked, 2).await.into_values() { + create.send(true).unwrap(); + } + timeout(WAIT, stop).await.unwrap().unwrap().unwrap(); + assert_eq!(0, failing_stores()); + assert_eq!( + vec![release(region_id, id(1, 1))], + releases_of(io.as_ref(), 2).await + ); + assert_eq!( + Counters { + created_objects: 2, + seal_to_durable: 2, + create_failures: 2, + sealed_objects: 2, + object_bytes: one_entry_object() + release_object(), + object_entries: 1, + release_records: 1, + rejected_appends: 3, + ..Counters::default() + }, + Counters::since(before) + ); + let store = open(object_store, &eager()).await; + assert_eq!(Some(id(1, 1)), watermark(&store, region_id)); + store.stop().await.unwrap(); + } + + /// Another writer holds the sequence: the release fails with the + /// conflict, which poisons the store, in both modes. + async fn release_create_finds_another_writer() { + for config in [eager(), enqueued(eager())] { + let object_store = memory_store(); + let (store, io, mut parked) = + open_parking_creates(object_store.clone(), &config).await; + put_foreign(&object_store, 1, 2).await; + let before = Counters::sample(); + let region_id = region(1); + let released = spawn_release(&store, region_id); + let (_, create) = next_parked_operation(&mut parked).await; + create.send(true).unwrap(); + let error = timeout(WAIT, released).await.unwrap().unwrap().unwrap_err(); + let path = io.object_path(1); + assert!( + matches!(unwrap_shared(&error), Error::WalObjectConflict { path: actual, .. } if *actual == path), + "unexpected error: {error:?}" + ); + assert_read_is_terminal(&store, region_id).await; + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + create_conflicts: 1, + poisoned: 1, + sealed_objects: 1, + object_bytes: release_object(), + ..Counters::default() + }, + Counters::since(before) + ); + let stopped = store.stop().await; + if config.ack_mode == AckMode::Durable { + stopped.unwrap(); + } else { + assert!(matches!( + unwrap_shared(&stopped.unwrap_err()), + Error::WalObjectConflict { .. } + )); + } + } + } + /// Appends over `io` after `arm` made the create meet the very object /// it writes, and asserts it was accepted as an identical retry. async fn assert_identical_retry(io: Arc, arm: impl FnOnce()) { @@ -10254,15 +11126,17 @@ mod tests { } /// Round-trips a command through the actor once `stop` began, when - /// the seal command is refused: the collection that `obsolete_all` of - /// a region without entries requests is answered by the actor, and + /// the seal command is refused: the actor refuses the release that + /// `obsolete_all` of a region without entries requests as well, which /// hides nothing of the regions a case reads. async fn round_trip_stopping_actor(store: &ObjectStoreLogStore) { let region_id = region(9); - store - .obsolete_all(&provider(region_id), region_id) - .await - .unwrap(); + assert_stopped( + &store + .obsolete_all(&provider(region_id), region_id) + .await + .unwrap_err(), + ); } /// Four creates are in flight and a fifth batch waits for a slot when diff --git a/src/metric-engine/Cargo.toml b/src/metric-engine/Cargo.toml index 92000549832..0cc39dc5add 100644 --- a/src/metric-engine/Cargo.toml +++ b/src/metric-engine/Cargo.toml @@ -49,6 +49,7 @@ common-function = { workspace = true } common-meta = { workspace = true, features = ["testing"] } common-wal = { workspace = true } criterion = { workspace = true, features = ["async", "async_tokio"] } +log-store = { workspace = true, features = ["testing"] } mito2 = { workspace = true, features = ["test"] } mur3 = "0.1" partition = { workspace = true } diff --git a/src/metric-engine/src/engine/drop.rs b/src/metric-engine/src/engine/drop.rs index 0b3d876fd84..7e98effdeee 100644 --- a/src/metric-engine/src/engine/drop.rs +++ b/src/metric-engine/src/engine/drop.rs @@ -171,3 +171,159 @@ impl MetricEngineInner { Ok(0) } } + +#[cfg(test)] +mod tests { + use api::v1::Rows; + use common_wal::config::object_store::{ObjectStoreWalConfig, STANDALONE_GENERATION}; + use common_wal::options::{ObjectStoreWalOptions, WAL_OPTIONS_KEY, WalOptions}; + use log_store::ObjectStoreLogStore; + use mito2::config::MitoConfig; + use mito2::test_util::TestEnv as MitoTestEnv; + use object_store::ObjectStore; + use object_store::services::Memory; + use store_api::logstore::LogStore; + use store_api::region_engine::RegionEngine; + use store_api::region_request::{RegionDropRequest, RegionPutRequest, RegionRequest}; + use store_api::storage::RegionId; + + use crate::test_util::{TestEnv, build_rows, row_schema_with_tags}; + use crate::utils::{to_data_region_id, to_metadata_region_id}; + + fn drop_request() -> RegionRequest { + RegionRequest::Drop(RegionDropRequest { + fast_path: false, + force: false, + partial_drop: false, + }) + } + + /// How the WAL releases of the dropped regions fail. + #[derive(Clone, Copy, PartialEq)] + enum ReleaseFailure { + /// The release of the data region fails transiently. + Data, + /// The releases of the data and the metadata region fail transiently. + Both, + /// The release of the data region meets a conflicting object, which + /// poisons the WAL, so the cleanup of the metadata region fails too. + Conflict, + } + + #[tokio::test] + async fn test_drop_physical_region_whose_wal_release_fails() { + for failure in [ + ReleaseFailure::Data, + ReleaseFailure::Both, + ReleaseFailure::Conflict, + ] { + let object_store = ObjectStore::new(Memory::default()).unwrap(); + let config = ObjectStoreWalConfig::default(); + let open_wal = || { + ObjectStoreLogStore::try_new( + object_store.clone(), + &config, + 0, + STANDALONE_GENERATION, + ) + }; + let wal = open_wal().await.unwrap(); + let mut mito_env = MitoTestEnv::with_prefix("metric-drop-failed-release").await; + let mito = mito_env + .create_engine_with_log_store(MitoConfig::default(), wal.clone()) + .await; + let env = TestEnv::with_mito_engine(mito_env, mito); + let region_id = RegionId::new(1, 1); + let wal_options = WalOptions::ObjectStore(ObjectStoreWalOptions::new( + "wal/datanodes/0/epochs/0".to_string(), + )); + env.create_physical_region( + region_id, + "test_metric_table", + vec![( + WAL_OPTIONS_KEY.to_string(), + serde_json::to_string(&wal_options).unwrap(), + )], + ) + .await; + // Both mito regions write to the WAL: the data region the rows of a + // logical region, the metadata region its creation and its drop. + let logical_region_id = RegionId::new(1024, 1); + env.create_logical_region(region_id, logical_region_id) + .await; + let metric = env.metric(); + metric + .handle_request( + logical_region_id, + RegionRequest::Put(RegionPutRequest { + skip_wal: false, + rows: Rows { + schema: row_schema_with_tags(&["job"]), + rows: build_rows(1, 10), + }, + hint: None, + partition_expr_version: None, + }), + ) + .await + .unwrap(); + metric + .handle_request(logical_region_id, drop_request()) + .await + .unwrap(); + + // The drop still removes both regions and the physical region state. + match failure { + ReleaseFailure::Data => wal.fail_next_create_after_write(), + ReleaseFailure::Both => wal.fail_creates(), + ReleaseFailure::Conflict => { + // A copy of the last object, of the same epoch, takes the + // sequence of the next one. + let objects = "wal/datanodes/0/epochs/0/objects/"; + let last = object_store + .list(objects) + .await + .unwrap() + .iter() + .filter_map(|entry| entry.name().strip_suffix(".wal")?.parse::().ok()) + .max() + .unwrap(); + let path = |object_seq: u64| format!("{objects}{object_seq:020}.wal"); + let copy = object_store.read(&path(last)).await.unwrap(); + object_store.write(&path(last + 1), copy).await.unwrap(); + } + } + metric + .handle_request(region_id, drop_request()) + .await + .unwrap(); + assert!(!env.mito().is_region_exists(to_data_region_id(region_id))); + assert!( + !env.mito() + .is_region_exists(to_metadata_region_id(region_id)) + ); + assert!( + !metric + .inner + .state + .read() + .unwrap() + .exist_physical_region(to_data_region_id(region_id)) + ); + + // The stop writes the release records of failed releases. A store + // opened later finds both regions released and keeps no segment of + // either. A poisoned store writes nothing more. + if failure == ReleaseFailure::Conflict { + assert!(wal.list_namespaces().await.is_err()); + continue; + } + wal.allow_creates(); + wal.stop().await.unwrap(); + let wal = open_wal().await.unwrap(); + wal.wait_for_garbage_collection().await; + assert!(wal.list_namespaces().await.unwrap().is_empty()); + wal.stop().await.unwrap(); + } + } +} diff --git a/src/metric-engine/src/test_util.rs b/src/metric-engine/src/test_util.rs index cf64d968182..3c3762aeea7 100644 --- a/src/metric-engine/src/test_util.rs +++ b/src/metric-engine/src/test_util.rs @@ -92,6 +92,17 @@ impl TestEnv { } } + /// Returns a new env whose metric engine runs over `mito`, an engine of + /// `mito_env`. + pub fn with_mito_engine(mito_env: MitoTestEnv, mito: MitoEngine) -> Self { + let metric = MetricEngine::try_new(mito.clone(), EngineConfig::default()).unwrap(); + Self { + mito_env, + mito, + metric, + } + } + /// Returns a new env with specific `prefix` and `mito_env` for test. pub async fn with_mito_env(mut mito_env: MitoTestEnv) -> Self { let mito = mito_env.create_engine(MitoConfig::default()).await; 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 54626b24432..9ecfc998697 100644 --- a/src/mito2/src/engine/object_store_wal_recovery_test.rs +++ b/src/mito2/src/engine/object_store_wal_recovery_test.rs @@ -1126,23 +1126,10 @@ async fn test_drop_cancels_a_flush_waiting_for_wal_durability() { wait_for(store.wait_for_durability_waits(1)).await; assert!(!flush.is_finished()); - // The drop cancels the flush instead of waiting behind the upload. - tokio::time::timeout( - WAIT, - engine.handle_request( - REGION_A, - RegionRequest::Drop(RegionDropRequest { - fast_path: false, - force: false, - partial_drop: false, - }), - ), - ) - .await - .unwrap() - .unwrap(); - assert!(!engine.is_region_exists(REGION_A)); - assert_eq!(vec![0], wal_object_seqs(&object_store).await); + // The drop cancels the flush instead of waiting behind the upload. Its + // release of the region then waits for its own object, which is created + // after the upload. + let dropped = spawn_drop(&engine, REGION_A); assert!( tokio::time::timeout(WAIT, flush) .await @@ -1150,8 +1137,17 @@ async fn test_drop_cancels_a_flush_waiting_for_wal_durability() { .unwrap() .is_err() ); + wait_for(store.wait_for_parked_creates(2)).await; + assert!(!dropped.is_finished()); + assert_eq!(vec![0], wal_object_seqs(&object_store).await); store.release_creates(); + tokio::time::timeout(WAIT, dropped) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!(!engine.is_region_exists(REGION_A)); tokio::time::timeout(WAIT, seal) .await .unwrap() @@ -1159,6 +1155,106 @@ async fn test_drop_cancels_a_flush_waiting_for_wal_durability() { .unwrap(); } +fn spawn_drop( + engine: &MitoEngine, + region_id: RegionId, +) -> tokio::task::JoinHandle> { + let engine = engine.clone(); + tokio::spawn(async move { + engine + .handle_request( + region_id, + RegionRequest::Drop(RegionDropRequest { + fast_path: false, + force: false, + partial_drop: false, + }), + ) + .await + .map(|_| ()) + }) +} + +#[rstest] +#[case::graceful(false, false)] +#[case::abrupt(true, false)] +#[case::failed_release(false, true)] +#[tokio::test] +async fn test_restart_after_a_drop_collects_the_objects_of_the_dropped_region( + #[case] abrupt: bool, + #[case] release_fails: bool, +) { + let mut env = TestEnv::with_prefix("object-store-wal-drop-restart").await; + let object_store = memory_store(); + let store = open_store(&object_store, AckMode::Durable).await; + let engine = new_engine(&mut env, store.clone()).await; + let (_, schema_a) = create_region(&engine, REGION_A, &[]).await; + let (table_dir_b, schema_b) = create_region(&engine, REGION_B, &[]).await; + + // Object 1 holds entries of both regions and object 2 of region A. The + // drop releases region A in object 3, and region B, which is not + // flushed, holds the boundary at object 1. + let mut writer = SealedWriter::new(&store); + writer + .put_and_seal( + &engine, + vec![ + (REGION_A, rows(&schema_a, 0, 2)), + (REGION_B, rows(&schema_b, 0, 3)), + ], + ) + .await; + writer + .put_and_seal(&engine, vec![(REGION_A, rows(&schema_a, 2, 4))]) + .await; + if release_fails { + store.fail_creates(); + } + let dropped = tokio::time::timeout(WAIT, spawn_drop(&engine, REGION_A)) + .await + .unwrap() + .unwrap(); + // A drop whose release fails succeeds all the same, and the stop creates + // the record in object 4. + dropped.unwrap(); + assert!(!engine.is_region_exists(REGION_A)); + store.allow_creates(); + wait_for_collection(&store).await; + let released_in = if release_fails { 4 } else { 3 }; + drop(writer); + if abrupt { + crash(engine, store).await; + } else { + engine.stop().await.unwrap(); + drop(engine); + store.stop().await.unwrap(); + drop(store); + } + assert_eq!( + vec![1, 2, released_in], + wal_object_seqs(&object_store).await + ); + + // Nothing is written before the restart, and region A is never opened + // again. + let store = open_store(&object_store, AckMode::Durable).await; + let engine = new_engine(&mut env, store.clone()).await; + open_region(&engine, REGION_B, &table_dir_b, &[]) + .await + .unwrap(); + + // Once region B is flushed no object holds a live segment, and the next + // object carries the boundary past all of them. + flush_region(&engine, REGION_B, None).await; + let mut writer = SealedWriter::new(&store); + writer + .put_and_seal(&engine, vec![(REGION_B, rows(&schema_b, 3, 5))]) + .await; + wait_for_collection(&store).await; + assert_eq!(vec![released_in + 2], wal_object_seqs(&object_store).await); + assert_eq!(5, engine.get_region_statistic(REGION_B).unwrap().num_rows); +} + #[rstest] #[case(RegionTruncateRequest::All)] #[case(RegionTruncateRequest::Unflushed)] diff --git a/src/mito2/src/test_util.rs b/src/mito2/src/test_util.rs index d5104bb16b7..f265494058a 100644 --- a/src/mito2/src/test_util.rs +++ b/src/mito2/src/test_util.rs @@ -495,7 +495,7 @@ impl TestEnv { } /// Creates a new engine with specific config on `log_store` under this env. - pub(crate) async fn create_engine_with_log_store( + pub async fn create_engine_with_log_store( &mut self, config: MitoConfig, log_store: Arc, diff --git a/src/mito2/src/wal.rs b/src/mito2/src/wal.rs index 76b3e3a2cdf..9d00a76d87f 100644 --- a/src/mito2/src/wal.rs +++ b/src/mito2/src/wal.rs @@ -220,6 +220,18 @@ impl Wal { .context(DeleteWalSnafu { region_id }) } + /// Releases every WAL entry of a dropped region, see [LogStore::release_region]. + pub async fn release_region(&self, region_id: RegionId, provider: &Provider) -> Result<()> { + if let Provider::Noop = provider { + return Ok(()); + } + self.store + .release_region(provider, region_id) + .await + .map_err(BoxedError::new) + .context(DeleteWalSnafu { region_id }) + } + /// Marks all WAL entries of a region as obsolete and removes its dedicated namespace when /// supported by the backend. pub async fn obsolete_all(&self, region_id: RegionId, provider: &Provider) -> Result<()> { diff --git a/src/mito2/src/worker/handle_drop.rs b/src/mito2/src/worker/handle_drop.rs index a62e8f0068e..192106110eb 100644 --- a/src/mito2/src/worker/handle_drop.rs +++ b/src/mito2/src/worker/handle_drop.rs @@ -121,14 +121,30 @@ where // this cleanup and close/drop remove. self.cleanup_dropped_region_runtime_state(region_id).await; - // Delete region data in WAL. - self.wal + // Deletes region data in WAL, and releases it so that this outlives a restart, which + // never opens the region again. A failure of the object store WAL does not fail the + // drop: the region has left the map, a drop request for it is not sent again, and a + // caller that drops several regions on one store, such as the metric engine, would + // leave the others undone, for instance once the release of one poisoned the store. + // The store keeps a failed release and writes it with its next object unless it is + // stopped or poisoned. + if let Err(e) = self + .wal .obsolete( region_id, region.version_control.current().last_entry_id, ®ion.provider, ) - .await?; + .await + { + if region.provider.as_object_store_provider().is_none() { + return Err(e); + } + error!(e; "Failed to obsolete the WAL of dropped region {}", region_id); + } + if let Err(e) = self.wal.release_region(region_id, ®ion.provider).await { + error!(e; "Failed to release the WAL of dropped region {}", region_id); + } if let Some(store) = &self.series_index_store { crate::series_index::delete_catalogs(store, region_id).await; diff --git a/src/store-api/src/logstore.rs b/src/store-api/src/logstore.rs index dd71142def5..ef0fbe49f01 100644 --- a/src/store-api/src/logstore.rs +++ b/src/store-api/src/logstore.rs @@ -106,6 +106,22 @@ pub trait LogStore: Send + Sync + 'static + std::fmt::Debug { ) -> Result<(), Self::Error> { Ok(()) } + + /// Releases every entry of a region the caller dropped, after `obsolete` + /// marked them obsolete, so that the log store keeps none of them even + /// across a restart, which never opens the region again to mark its + /// entries obsolete. + /// + /// The default does nothing and keeps each log store's existing handling + /// of a dropped region. A log store that keeps obsolete marks only in + /// memory overrides this method and returns once the release is durable. + async fn release_region( + &self, + _provider: &Provider, + _region_id: RegionId, + ) -> Result<(), Self::Error> { + Ok(()) + } } /// The response of an `append` operation. diff --git a/tests/compatibility/README.md b/tests/compatibility/README.md index 404b6332b94..1daf42cc736 100644 --- a/tests/compatibility/README.md +++ b/tests/compatibility/README.md @@ -15,6 +15,12 @@ read-preference defaulting and unrelated metadata preservation: cargo nextest run -p session -E 'test(test_legacy_query_context_wire_defaults_to_leader)' ``` +## Object store WAL fixtures + +Frozen objects of the object store WAL are checked byte for byte by the format tests of +`log-store`. The SQL runner cannot cover the object store WAL: it owns `wal.provider`. See the +[fixture notes](wal/object_store/README.md). + ## Quick Start ```shell diff --git a/tests/compatibility/wal/object_store/README.md b/tests/compatibility/wal/object_store/README.md new file mode 100644 index 00000000000..411900197f4 --- /dev/null +++ b/tests/compatibility/wal/object_store/README.md @@ -0,0 +1,16 @@ +# Object store WAL object fixtures + +Frozen objects of the object store WAL, checked byte for byte by the format tests in +`src/log-store/src/object_store_wal/format.rs`, which never rewrite them: + +```shell +cargo nextest run -p log-store -E 'test(test_format_matches_release_record_fixture)' +``` + +## `release_record.wal` + +A version 1 object, sequence 3, holding entry `(3 << 20) | 1` of region 1 and, as the single +entry of the reserved region `u64::MAX` at the same position, the release records of region +`(2, 1)` through `(2 << 20) | 1` and of region `(3, 0)` in full. A reader that predates +release records decodes it as an ordinary version 1 object. 272 bytes, SHA-256 +`7e8de4d50142f3b48c97a24f20744ee9c0d56f843daf3a36d00ba9b465a5d4b5`. diff --git a/tests/compatibility/wal/object_store/release_record.wal b/tests/compatibility/wal/object_store/release_record.wal new file mode 100644 index 00000000000..0318e3630d1 Binary files /dev/null and b/tests/compatibility/wal/object_store/release_record.wal differ