neuwirth
/
LeVCS
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Answer a retry of a committed operation 

A client that submitted, lost the answer, and submitted again was refused.
Retrying is the normal case, and the operation ID exists precisely so the
store can say what happened the first time instead of either appending a
second frame or turning a recoverable disconnect into a failure.

The rule is `oracle::coalescing_decision`'s durable branch applied to a
terminal entry rather than restated beside it: a matching stable digest
returns the durable receipt, and a different one is a conflict. The digest
is what makes the answer safe. An operation ID alone cannot tell 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.

An expired entry is refused rather than answered. Its tombstone still
binds the ID against reuse, but the receipt is gone and a tombstone is not
a statement about this submit's outcome. The remaining `TransactionStatus`
variants are named and poison: a terminal entry reporting `Pending`,
`Resolving`, or `Unknown` is a contradiction, and a catch-all would answer
it with whatever the last arm happened to be.

The receipt reaches `accept` as a field rather than through `StoreError`,
so `accept` completes the waiter with it. Carrying a success through the
error channel would make every caller of `prepare` responsible for
noticing that one variant means it worked.

Nothing is sequenced and no frame is appended for a retry, and the tests
replay one through a reopen, where the answer can only come from the root
recovery rebuilt.

In-flight coalescing -- a same-digest resubmit attaching to a leader that
has not resolved yet -- is still refused. It changes waiter and completion
ownership and is left to its own change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JGdH5V43XWnj1PdHqiPktQ
This commit is contained in:
Levi Neuwirth 2026-08-09 16:23:22 +02:00
parent 9b2117a406
commit d884db3f3e
No known key found for this signature in database
2 changed files with 279 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
crates/levcs-store/src/engine.rs
View File

@ -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<CommitReceipt>,
/// 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

208
crates/levcs-store/tests/idempotency.rs Normal file
View File

@ -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()
)
});
}