diff --git a/crates/levcs-store/src/engine.rs b/crates/levcs-store/src/engine.rs index 00e5828..4ca32b4 100644 --- a/crates/levcs-store/src/engine.rs +++ b/crates/levcs-store/src/engine.rs @@ -1543,6 +1543,7 @@ fn spawn_shard_writer( repositories, tail_generation: tail.logical_generation, rotation_wanted: false, + durable_answer: None, run_entry_baseline, checkpoint_namespaces, journal, @@ -1624,6 +1625,10 @@ struct ShardWriter { /// Logical index generation of the active journal, so an index location /// built here resolves to the same pinned file after a reopen. tail_generation: u64, + /// Set by `prepare` when the operation already committed and the caller is + /// retrying. Carries the durable receipt out to `accept`, which completes + /// the waiter with it -- a retry is answered, not refused. + durable_answer: Option, /// Set by `sequence_into_frame` when the forming group no longer fits the /// active journal, and consumed by `accept`. A flag rather than an error /// variant because the condition is not a refusal — nothing is wrong with @@ -1871,6 +1876,19 @@ impl ShardWriter { let (mut prepared, frame) = match self.prepare_caught(&transaction) { Ok(pair) => pair, Err(error) => { + // A retry of an operation that already committed. The receipt + // is the durable one, so nothing is sequenced and no frame is + // appended for it. + if let Some(receipt) = self.durable_answer.take() { + drop(adoption); + if let Some(waiter) = self.waiters.pop() { + if let Some(key) = waiter.reservation { + self.release_reservation(&key); + } + waiter.completion.complete(Ok(receipt)); + } + return; + } if !std::mem::take(&mut self.rotation_wanted) { refuse!(error); } @@ -2127,13 +2145,59 @@ impl ShardWriter { // Decision 9.8 ordering, first rule: a committed or expired operation // is answered from the committed root and never touches the status - // root. The answer itself — returning the durable receipt to a - // retrying caller — is deliverable 7. - if self.shared.committed.load().terminal_status(&key).is_some() { - return Err(StoreError::NotImplemented( - "submit idempotency against an already committed operation — B1 NamespaceTxn, \ - scope 6-B1 deliverable 7", - )); + // root. + // + // The rule is `oracle::coalescing_decision`'s durable branch — + // `ReturnDurable` on a matching digest, `OperationIdMismatch` + // otherwise — applied to a terminal entry rather than restated. What + // makes the digest load-bearing is that an operation ID alone cannot + // distinguish a retry from a different request reusing an identity, and + // answering the second with the first's receipt would tell a caller its + // transaction committed when another one did. + if let Some(entry) = self.shared.committed.load().terminal_status(&key) { + if entry.operation_digest() != transaction.operation_digest { + return Err(StoreError::Conflict(format!( + "operation {} already resolved under a different stable digest; one \ + operation ID names one request", + key.operation_id.to_hex() + ))); + } + return match entry.transaction_status() { + // Not an error, so it travels as a flag rather than through the + // error channel: `accept` completes this waiter with the + // receipt instead of failing it. Smuggling a success through + // `StoreError` would make every caller of `prepare` responsible + // for noticing that one variant means "it worked". + TransactionStatus::Committed(receipt) => { + self.durable_answer = Some(receipt); + Err(StoreError::Overloaded { + limit: "durable_answer", + retry_after_micros: 0, + }) + } + // The receipt aged out and only the tombstone remains, which is + // what stops the ID being reused — but it is no longer an + // answer about *this* submit's outcome, so it cannot be + // returned as one. + TransactionStatus::Expired { + tombstone_until_micros, + .. + } => Err(StoreError::Conflict(format!( + "operation {} committed and its receipt is no longer retained; the ID stays \ + bound until {tombstone_until_micros}", + key.operation_id.to_hex() + ))), + // A terminal entry is `Committed` or `Expired` by construction. + // Naming the rest is the no-catch-all rule: a third terminal + // shape must fail this compile rather than fall through to a + // default answer about a durable operation. + status @ (TransactionStatus::Pending { .. } + | TransactionStatus::Resolving { .. } + | TransactionStatus::Unknown) => Err(self.poison_error(format!( + "operation {} has a terminal entry reporting {status:?}", + key.operation_id.to_hex() + ))), + }; } // Plan §5.3: at a hard ceiling the shard seals synchronously **before diff --git a/crates/levcs-store/tests/idempotency.rs b/crates/levcs-store/tests/idempotency.rs new file mode 100644 index 0000000..0474996 --- /dev/null +++ b/crates/levcs-store/tests/idempotency.rs @@ -0,0 +1,208 @@ +//! Scope 6-B1 deliverable 7 — resubmitting an operation that already committed. +//! +//! A retry is the normal case, not an error. A client that submitted, lost the +//! connection, and submitted again must be told what happened the first time +//! rather than either appending a second frame or being refused. The operation +//! ID is the identity that makes that answerable, and the stable digest is what +//! makes the answer safe: the same ID over the same digest is the same request, +//! and the same ID over a *different* digest is two different requests claiming +//! one identity, which is the case that must be refused. +//! +//! Asserted against `oracle::coalescing_decision`, which is the frozen +//! statement of that rule, rather than against a second copy of it written +//! here. The oracle is fed the same three inputs the store has -- the durable +//! status, the in-flight digest, and the incoming digest -- and the store's +//! answer must be the one it names. + +#![cfg(all( + feature = "store-privileged", + feature = "store-internals", + feature = "failpoints" +))] + +use levcs_core::{ObjectId, ObjectType}; +use levcs_protocol::oracle::{coalescing_decision, CoalescingDecision}; +use levcs_protocol::v2::TransactionStatusV1; +use levcs_store::transaction::StagedObject; +use levcs_store::types::{NamespaceId, OperationId, PrivilegedConstruction, StoreError}; +use levcs_store::ValidatedTransaction; + +#[path = "support/engine_matrix.rs"] +mod engine_matrix; + +use engine_matrix::{ + create_transaction, deadline, evidence, genesis_id, namespace_on_shard, open_absent_root, + reopen_after_close, submit, DEFAULT_MAX_INDEX_RUNS, +}; + +const SHARD_COUNT: u16 = 1; +const OPERATION: u8 = 0x40; + +/// One transaction, parameterized by the payload that determines its stable +/// digest, so the same operation ID can be resubmitted with the same digest or +/// a different one. +fn transaction(namespace: NamespaceId, digest_seed: u8, blob: u8) -> ValidatedTransaction { + let authority = genesis_id(&namespace); + ValidatedTransaction::builder(PrivilegedConstruction::assert_validated()) + .namespace(namespace) + .operation( + OperationId([OPERATION; 16]), + ObjectId([digest_seed; 32]), + deadline(), + ) + .objects(vec![StagedObject { + id: ObjectId([blob; 32]), + object_type: ObjectType::Blob, + raw: vec![blob; 64], + }]) + .refs(Vec::new()) + .authority(Some(authority), Some(authority)) + .evidence(evidence()) + .build() + .expect("a complete transaction") +} + +/// The durable status the oracle is asked about, built from the receipt the +/// store actually produced. Only `operation_digest` participates in the +/// decision; the rest is carried so the value is a real status rather than a +/// shape that happens to satisfy one accessor. +fn durable_status( + receipt: &levcs_store::types::CommitReceipt, + digest: ObjectId, +) -> TransactionStatusV1 { + TransactionStatusV1::Committed(levcs_protocol::v2::CommitReceiptV1 { + operation_id: *receipt.operation_id.as_bytes(), + operation_digest: digest, + repo_sequence: receipt.repo_sequence, + current_authority: receipt.current_authority, + refs: Vec::new(), + objects_new: receipt.objects_new, + retry_until_micros: 0, + first_visible_at_micros: 0, + receipt_visible_until_micros: 0, + }) +} + +#[test] +fn a_resubmitted_operation_returns_its_durable_receipt() { + let directory = tempfile::TempDir::new().expect("a temporary root"); + let namespace = namespace_on_shard(0, SHARD_COUNT, 1); + let engine = open_absent_root(directory.path(), SHARD_COUNT, DEFAULT_MAX_INDEX_RUNS); + submit(&engine, create_transaction(namespace, 1)) + .receipt() + .expect("the repository is created"); + + let first = submit(&engine, transaction(namespace, 0xd1, 0xb1)); + let first = first + .receipt() + .unwrap_or_else(|| panic!("the first submit must commit: {:?}", first.error())) + .clone(); + + // The oracle's answer for these inputs, stated before the store is asked. + assert!( + matches!( + coalescing_decision( + durable_status(&first, ObjectId([0xd1; 32])), + None, + ObjectId([0xd1; 32]) + ), + CoalescingDecision::ReturnDurable(_) + ), + "the oracle must name this a durable return, or this test is asserting the wrong rule" + ); + + let again = submit(&engine, transaction(namespace, 0xd1, 0xb1)); + let again = again.receipt().unwrap_or_else(|| { + panic!( + "a same-digest resubmit must be answered: {:?}", + again.error() + ) + }); + assert_eq!( + again.repo_sequence, first.repo_sequence, + "a resubmit must return the receipt the first submit produced, not a new one; a second \ + repo_sequence means a second frame was appended for one operation" + ); + assert_eq!(again.objects_new, first.objects_new); + + // And after a reopen, where the answer can only come from the committed + // root recovery rebuilt. + drop(engine); + let reopened = reopen_after_close(directory.path(), SHARD_COUNT, DEFAULT_MAX_INDEX_RUNS); + let recovered = submit(&reopened, transaction(namespace, 0xd1, 0xb1)); + let recovered = recovered.receipt().unwrap_or_else(|| { + panic!( + "a resubmit after reopen must still be answered: {:?}", + recovered.error() + ) + }); + assert_eq!( + recovered.repo_sequence, first.repo_sequence, + "the durable answer must survive recovery" + ); +} + +#[test] +fn one_operation_id_with_a_second_digest_is_refused() { + let directory = tempfile::TempDir::new().expect("a temporary root"); + let namespace = namespace_on_shard(0, SHARD_COUNT, 1); + let engine = open_absent_root(directory.path(), SHARD_COUNT, DEFAULT_MAX_INDEX_RUNS); + submit(&engine, create_transaction(namespace, 1)) + .receipt() + .expect("the repository is created"); + + let first = submit(&engine, transaction(namespace, 0xd1, 0xb1)); + let first = first + .receipt() + .unwrap_or_else(|| panic!("the first submit must commit: {:?}", first.error())) + .clone(); + + assert!( + matches!( + coalescing_decision( + durable_status(&first, ObjectId([0xd1; 32])), + None, + ObjectId([0xd2; 32]) + ), + CoalescingDecision::OperationIdMismatch + ), + "the oracle must name this a mismatch, or this test is asserting the wrong rule" + ); + + // Same operation ID, different stable digest: two different requests + // claiming one identity. Returning the first receipt would tell the caller + // its transaction committed when a different one did. + let conflicting = submit(&engine, transaction(namespace, 0xd2, 0xb2)); + match conflicting.error() { + Some(StoreError::Conflict(message)) => assert!( + message.contains("digest"), + "the refusal must say what disagreed: {message}" + ), + other => panic!("a same-ID/different-digest resubmit must be refused, got {other:?}"), + } + + // The refusal is about identity, not about the shard: an unrelated + // operation still commits. + let unrelated = submit(&engine, { + let authority = genesis_id(&namespace); + ValidatedTransaction::builder(PrivilegedConstruction::assert_validated()) + .namespace(namespace) + .operation(OperationId([0x77; 16]), ObjectId([0x77; 32]), deadline()) + .objects(vec![StagedObject { + id: ObjectId([0x77; 32]), + object_type: ObjectType::Blob, + raw: vec![0x77; 64], + }]) + .refs(Vec::new()) + .authority(Some(authority), Some(authority)) + .evidence(evidence()) + .build() + .expect("a complete transaction") + }); + unrelated.receipt().unwrap_or_else(|| { + panic!( + "an unrelated operation must commit: {:?}", + unrelated.error() + ) + }); +}