Repository navigation
fix(cassandra): record stream records for transacted writes - #576
Open
LeeroyHannigan wants to merge 9 commits into
Open
LeeroyHannigan wants to merge 9 commits into
LeeroyHannigan wants to merge 9 commits into
Conversation
A TransactWriteItems Put, Update or Delete on a Cassandra table with an enabled stream recorded nothing. The commit phase wrote the item mutation and such index rows as it maintained, and skipped the stream capture the request carries per operation. The commit phase now builds the stream record for each item mutation with the same stream_util::stream_record_statement the single-item paths use, with the same image pair, so the event name, the images for the table's view type, the shard assignment and the sequence number come from the same source as the equivalent PutItem, UpdateItem or DeleteItem. The record joins the secondary-effects LOGGED BATCH that already carries the index rows; Cassandra rejects a batch mixing an LWT with writes to another partition, so neither the single-item path nor this one can put the record in the base mutation's own statement. A pure builder returns that batch and the commit path executes what it returns, so a unit test holds the index rows and the record together. The old image is read at commit time, from the row together with its created_to_prepare and last_committed_txn_timestamp columns: a row PREPARE planted, which has never had a commit land on it, has no old image; otherwise the old image is the row's item_data, which is itself absent when the row holds a marker with no item. PREPARE plants the request's own item for a Put and the bare key for an Update, so the row's item_data alone would report an insert as a modify, and PREPARE's own read is one write stale when a single-item write commits inside the PREPARE window, where it is accepted and the marker's IF prepared_txn_id = null does not detect it. This costs one read per streamed Put and Delete op; the Update arm reads the row anyway. Committing, and the recovery worker's replay, now clear created_to_prepare. Releases before this one did not clear it, so a row they committed carries the flag over a committed item; the commit timestamp, which only a commit writes, tells that row apart from a planted one, and the first transacted write on it after an upgrade records MODIFY or REMOVE rather than INSERT or nothing. The index metadata read moves above the base LWT on the Put and Update arms, as occ_write does it, so the record does not wait an uncached catalog round trip behind an item that has already committed. A ConditionCheck records nothing, a cancelled transaction records nothing for any of its items, and a delete of an absent key records nothing. Records land in request order, which is ExtendDB's own behaviour on every backend rather than a promise the service makes. Storage tests cover fourteen shapes on a live Cassandra node, including a Put over an existing item, a Put over a planted row, a Put and a Delete over an item committed before the flag was cleared, and a transacted record compared field by field with the single-item record of the same verb. A wire test covers a transacted Put over an existing item, an Update, a Delete, a ConditionCheck, a cancelled transaction and a KEYS_ONLY stream; it asserts per key, since records for one key share a shard and records for different keys do not. Closes #516
LeeroyHannigan
requested review from
amrith,
c33howard,
jcshepherd,
pdf-amzn,
robinnsc and
yesyayen
as code owners
October 25, 2026 20:45
This was referenced Oct 25, 2026
A transacted write on a table with a stream and an index put the stream record in the same LOGGED BATCH as the index rows. That batch runs after the base row's lightweight transaction has committed. Cassandra rejects a batch that writes more than one partition once its data passes batch_size_fail_threshold, which is 51,200 bytes unless the operator changed it. The rejection left the item written without its index rows or its record, left the ledger entry in COMMITTING, left the later operations of the transaction uncommitted, and answered 500. secondary_effect_batches now adds up the bytes of the index statements (text and bound values) and of the record statement. Up to 24,000 bytes the record joins the index batch as before. Past that, the index rows run first and the record runs after them as a batch of its own. A batch with one statement writes one partition and is not subject to the limit, so a record of any size commits. commit_secondary_effects already ran the batches in order. It now builds the index batch before calling the helper, because the size can only be read from a built batch. The budget is 24,000 bytes. On cassandra:4.1 with the default threshold, a table with one GSI and a NEW_AND_OLD_IMAGES stream took a transacted insert of 24 KB and refused one of 28 KB, and took a transacted overwrite of 16 KB and refused one of 20 KB. The index entry and the record total about 48 KB where those committed and 56 to 60 KB where they were refused, which fits the 51,200 byte limit. 24,000 is under half of the limit, so the margin covers the difference between this count and the size Cassandra measures, and a server configured with a lower threshold, which the code cannot read. The count includes the index statements because every index adds a copy of the item. An 18 KB insert on a table with two GSIs has a 19 KB record, which is under any cutoff on the record alone, and the record plus the two index entries is 57 KB. For the writes it applies to, the split means the index rows and the record are no longer one atomic unit. If the process stops after the first batch, or the second batch fails, the index rows are written and the record is not, and nothing writes the record afterwards. A storage test overwrites a 24 KB item in a transaction on a GSI table with a NEW_AND_OLD_IMAGES stream and requires success, one MODIFY record with both images, and no ledger entry. A second storage test runs an insert, an overwrite and a two-operation transaction under a client token at 16, 20, 24, 28 and 32 KB. Two more cover a transacted Update of a 20 KB item and the two-index insert above. Unit tests pin that a 30,000-byte record yields two batches with the record last, where the budget boundary lies, that bound values count toward it, and that a record with no index rows stays one batch whatever its size.
The transacted Put, Update and Delete arms built the stream record statement before the base write. The statement takes its sequence number when it is built, so the number was taken before the item's lightweight transaction. For a Put or Update it was also taken before the catalog read for the table's indexes, and for a Delete before two or three lightweight transactions. A record numbered earlier than the write it describes can land after a record that another write on the same shard has already published, and a consumer that has read the later record can miss the earlier one. The arms now read the old image before the base write and build the statement after it: after commit_put_or_update for a Put or Update, and after check_lwt_applied for a Delete. The old image is the same row as before. The Update arm already held it from the read it uses to apply the update. On loopback against cassandra:4.1, 40 single-operation transactions (20 Puts and 20 Updates) took a sequence number whose millisecond was earlier than WRITETIME(item_data) of the stored item in 25 cases on the previous commit, 14 of the Puts and 11 of the Updates, each by 1 ms. With this change it was 0 of 40. The new storage test makes that comparison for the same 20 Puts and 20 Updates. It does not cover the Delete arm, because a deleted row has no write time to compare with.
Committing a transacted Put or Update clears created_to_prepare as well as prepared_txn_id. The doc comment on commit_put_or_update gave a different reason, that a flag left set would make the next transacted write on the key record an insert over a committed item. The commit-timestamp rule in fetch_row_for_transaction already tells a committed row from a placeholder when the flag is set, so the clearing does not change which old image a record carries. The clearing affects rollback. rollback_single_operation and recover_rollback_op run DELETE .. IF prepared_txn_id = ? AND created_to_prepare = true before they clear prepared_txn_id. A committed row that kept the flag matches that DELETE when a later transaction that prepared the key is cancelled, and the committed item is deleted. The doc comment now says this. It also says that rows committed before the clearing existed still carry the flag and are still deleted this way. The new storage test commits an item through a transaction, then runs a transaction that overwrites the key and carries a condition on another key that fails. It requires the cancelled transaction to leave the first transaction's value in place and the other key unwritten. With the clearing taken out of both commit statements, get_item returns nothing after the cancelled transaction.
The scenario in test_transact_write_records_one_stream_record_per_mutation, a Put, an Update and a Delete in one transaction, now lives in a function that takes the table and the builders for its items and keys. The original test calls it on a table with a hash key only. New tests call it on tables with a string, a number and a binary sort key, checking that every record carries both key attributes, and on a table with a GSI pinned to synchronous propagation, checking that the Put's index row is written. Two more tests pin that the index rows and the record of a small item are one batch. In the first, the index's table has been removed, so the server rejects the index statement. The transaction fails with an error from the secondary-effects batch and no record exists. In the second, the account's stream_records table has been removed, so the server rejects the record statement. The transaction fails the same way and the index scan holds no row. The first test fails if the record can be written when the index statement is rejected, and the second if the index rows can be written when the record statement is rejected.
commit_secondary_effects said the queue row an asynchronous GSI takes is written outside the batch. enqueue_async_indexes adds it to the same batch as the index rows, and its own doc comment says so. The same comment called the window after the base LWT brief and compared it to the one async GSIs accept. An async GSI row converges through its queue. An index row or a record lost after the base LWT is not written again, because recovery re-runs the base mutation only. The comment now says that, and says the queue row goes into the index batch. indexes_for_commit said occ_write reads the indexes before its own base write. occ_write receives them from its callers, which read them once at entry, so the sentence is dropped. The planted-row helper in the images test described a row with no prepared_txn_id and called it the shape a single-item write leaves. PREPARE sets prepared_txn_id, and no single-item path writes created_to_prepare. The comment now describes the row PREPARE plants and what the helper writes. The comment in the transact_write_items engine handler said stream records are captured atomically within the storage transaction. That holds for PostgreSQL, SQLite and MongoDB, and not for Cassandra, and the comment now says which. The debug_assert_eq on pre_commit_images is removed. The loop above it pushes one image per op in every arm, so the assertion cannot fail. A docstring in the Python test now says "an extenddb endpoint".
The doc comment on secondary_effect_batches said that splitting the record into its own batch keeps a large write from reaching the state where the item is written without its index rows or its record. The split returns the index batch unchanged. Cassandra refuses any batch that writes more than one partition once its data passes batch_size_fail_threshold, 50 KiB by default, whatever statements the batch holds. On two synchronous GSIs, or one synchronous and one asynchronous GSI, the index rows alone pass that limit from an item of about 26 KB, with or without a stream, and the batch is refused after the base row's lightweight transaction has committed. The comment now says that the split keeps the record from being the statement that pushes a batch over the limit, that it does not bound the index batch, and when the item is written without its index rows. The sentence about the lost atomicity said "It" after the replaced sentence, so it now names the split. No code changes.
commit_put_or_update clears created_to_prepare in two statements, one for a table with a sort key and one for a table with a hash key only. The cancelled-transaction test created its table with no sort key, so every commit in it went through the hash-key statement and no test ran the sort-key statement's clearing. A committed sort-key row that kept the flag matches the rollback DELETE of a later cancelled transaction that prepared the key, and the committed item is deleted. The scenario now lives in a function that takes the table and the builders for its items and keys, and the existing test calls it on the hash-key table. A second test calls it on a table with a string sort key, with the sort-key item and key builders of the stream tables tests, whose with_sort helper is now visible to the other test modules. It requires the item the first transaction committed to be stored with its first value after a second transaction that prepared the same key is cancelled, and the second transaction's other key to be unwritten.
secondary_effect_batches builds the record's batch from the index batch's request and copies its batch type, consistency, serial consistency, timestamp, keyspace and now_in_seconds. The unit test for a record too large to share the index batch compared the consistency and the batch type only, and its index batch left the other four settings at their defaults, so a record batch that dropped them would have passed. The shared index_batch fixture now sets the serial consistency, the timestamp, the keyspace and now_in_seconds to values other than their defaults. The test checks that the index batch has all four set, then requires the record batch to carry each of them.
LeeroyHannigan
added a commit
that referenced
this pull request
Oct 27, 2026
The Cassandra pytest job failed two tests of tests/test_stream_write_contention.py on this branch. The batch test rejected a PutItem that answered 400 TransactionConflictException. Cassandra settles a lost optimistic-concurrency race that way once its retries are spent, and the UpdateItem test in the same file already accepts it when EXTENDDB_TEST_BACKEND names that backend. The batch test now accepts it the same way; MongoDB, SQLite and PostgreSQL keep 503 as the only allowed failure. The transaction test expected one stream record per operation of every transaction. Cassandra writes no stream record for a transacted write until PR #576 lands, so the stream held only the setup puts (60 of the 240 wanted). On that backend the test now expects the setup puts alone; the assertions that every transaction committed and that each counter holds its writer's rounds run unchanged on every backend.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #516.
What was wrong
On Cassandra a
TransactWriteItemsPut, Update or Delete on a table with an enabled stream recorded nothing;PutItem,UpdateItem,DeleteItemandBatchWriteItemon the same table recorded as expected, and SQLite recorded all five shapes. On amainbinary four of the five new wire tests fail; on the branch all pass on every backend.What changed
One source file,
crates/storage-cassandra/src/data/transactions.rs, plus a new siblingdata/transact_stream.rsholding the pure pieces (the old-image rule and the batch builder) with their unit tests.The commit phase builds a stream record for each item mutation with the same
stream_record_statementevery single-item path uses, with the same image pair. A pure builder decides how the record is batched with the index rows and the commit path executes what it returns. Unit tests pin how the statements are grouped. Two storage tests pin that a rejected index statement leaves no record and a rejected record statement leaves no index row.The stream record goes into the same LOGGED BATCH as the index rows, and that batch runs after the base row's lightweight transaction. Cassandra rejects a batch that writes more than one partition once its data passes
batch_size_fail_threshold, which is 51,200 bytes unless the operator changed it. A large item therefore made the transaction fail after its first operation had committed. On a table with one GSI and a NEW_AND_OLD_IMAGES stream, with the default threshold, a transacted insert committed at 24 KB and failed at 28 KB, and a transacted overwrite committed at 16 KB and failed at 20 KB. The failed transaction left the item written without its index rows or its record, left the ledger entry in COMMITTING, left the later operations uncommitted, and answered 500.secondary_effect_batchesnow adds up the bytes of the index statements and the record statement. Up to 24,000 bytes the record shares the index batch. Past that, the index rows run first and the record runs after them as a batch of its own. A batch with one statement writes one partition, so the limit does not apply to it. 24,000 is under half of the default limit, so the margin covers the difference between this count and the size Cassandra measures, and a server set to a lower threshold. The code cannot read that setting. The count includes the index statements because each index adds a copy of the item: an 18 KB insert on a table with two GSIs has a 19 KB record, and 57 KB with its two index entries.For the writes over the budget, the index rows and the record are no longer one atomic unit. If the process stops after the first batch, or the second batch fails, the index rows are written and the record is not, and nothing writes the record afterwards. With the split, an insert, an overwrite and a two-operation transaction at 16, 20, 24, 28 and 32 KB all commit with their records on the one-GSI table, where the single batch failed from 28 KB for an insert and from 20 KB for an overwrite. The split does not bound the index batch, which is sent whole: when the index rows write more than one partition and pass the limit by themselves (two synchronous GSIs, or one synchronous and one asynchronous GSI, from about 26 KB), the item is written without them. That failure is the same on
mainand is filed separately.The sequence number of each record is taken after the base write, on the Put, Update and Delete arms. Before this change it was taken before the lightweight transaction, and in 25 of 40 transacted Puts and Updates the record's millisecond was earlier than the row's own write time, so a consumer reading by sequence number could miss a record that landed behind a later-numbered one; after the change 0 of 40.
The old image is read at commit time from the row together with
created_to_prepareandlast_committed_txn_timestamp. PREPARE plants a placeholder row holding the request's own item for an absent key, so the row'sitem_dataalone would report an insert as a modify; and PREPARE's own read is one write stale when a single-item write commits inside the PREPARE window, whichmainaccepts and the marker does not detect. The commit and the recovery replay now clearcreated_to_prepare. Rows committed by earlier releases still carry the flag over a committed item; the commit timestamp, which only a commit writes, tells them apart, so the first transacted write on such a row after an upgrade records MODIFY or REMOVE rather than INSERT or nothing. Clearing the flag also means a committed item is kept when a later transaction that prepared the key is cancelled: the rollback deletes a row only when the flag is still set, and a storage test on a hash-key and on a sort-key table pins that the committed item survives.The index metadata read moved above the base write on the Put and Update arms, so the record does not wait an uncached catalog round trip behind an already-committed item.
Measured on Amazon DynamoDB (2026-10-25)
Record order within one transaction: the service does not return a transaction's records in request order (a Put to
k5then a Put tok7came backk7,k5; a Put, Update and Delete came back REMOVE, INSERT, MODIFY, one shard, composite key). ExtendDB returns request order on every backend. The wire tests assert per key, which is the guarantee the service gives.Record contents: a transacted record carries the same field set as a single-item record (
awsRegion,dynamodb,eventID,eventName,eventSource,eventVersion; insidedynamodb,ApproximateCreationDateTime,Keys, the images for the view type,SequenceNumber,SizeBytes,StreamViewType), nouserIdentity, nothing marking it as transacted; sequence numbers of one transaction are not contiguous. ExtendDB matches: same builder, same arguments, one number per mutation from the same clock. On Cassandra and SQLite the transacted record equals the single-item record of the same verb field for field, for a Put, an Update and a Delete under NEW_AND_OLD_IMAGES.Tests
Storage (
tests/direct/transact_write_stream_records.rs,transact_write_stream_images.rs,transact_write_stream_batch_limit.rs,transact_write_stream_size_sweep.rs,transact_write_stream_sequence.rs,transact_write_stream_tables.rs,transact_write_rollback_after_commit.rs, in thedirect_integrationbinary CI runs whole): one record per mutation, two mutations in request order, a streamed and an unstreamed table in one transaction,KEYS_ONLY,ConditionCheck, a cancelled transaction, a delete of an absent key, an update creating a key, a Put over an existing item, a Put over a planted row, a Put and a Delete over a row committed before the flag was cleared, a field-for-field comparison with the single-item records, the batch limit (a 24 KB overwrite under NEW_AND_OLD_IMAGES on a GSI table commits with one MODIFY record and no ledger entry; the size sweep at 16 to 32 KB), the sequence order (no record's millisecond earlier than its row's write time in 20 Puts and 20 Updates), the all-verbs test on S, N and B sort-key tables and on a GSI table, a failing index statement leaving no record and a failing record statement leaving no index row, and a cancelled transaction keeping a committed item on a hash-key and on a sort-key table. Unit: eight tests on the pure functions, including the budget boundary, bound values counting toward it, and the record batch carrying the index batch's type, consistency, serial consistency, timestamp, keyspace and now-in-seconds. Wire (tests/python/test_streams_transactions.py, collected by every backend job): five tests, 36 of 36 withtest_streams.pyandtest_transactions.pyon Cassandra, and the same on SQLite, PostgreSQL and MongoDB.Controls, each on a committed tree: the fix removed, 6 of 8 original storage tests and 4 of 5 wire tests fail; the old image taken from PREPARE's read instead of the commit-time read, only the planted-row test fails; the Put arm dropping its old image, two storage tests and the wire row fail; the record forced back into the index batch, the four batch-limit tests and the size sweep fail with the batch-too-large error after the lightweight transaction; the sequence number taken before the write, 9 to 14 of 40 records early; the
created_to_prepareclearing removed, the committed item is lost on the hash-key and on the sort-key table; the budget counting only the record, one integration test and two unit tests fail; the commit-timestamp rule removed, the two upgrade tests fail.cargo fmt, clippy 1.97 and the 1.88 build are clean; 69 lib tests;direct_integrationwhole 173 of 173 at four threads.Stated limits
A failure while writing op k's index rows or record leaves k items mutated and k-1 records written. Op k's index rows are missing too, unless the record was split into its own batch and the failure came in that second batch. A failure in the read or the lightweight transaction of op k+1 leaves k items mutated and k records written. The recovery worker finishes the base mutations of the remaining ops and writes no further records and no index rows. A transaction recovered after a crash writes no stream record and no index rows, because the ledger carries neither the capture nor the key schema. Both are the backend's pre-existing behaviour on this path.
A single-item write that commits inside a transaction's PREPARE window is accepted and not detected, and an item deleted there is resurrected by the transaction; pre-existing, filed separately. This branch's commit-time read means the record reflects the row as committed, so it no longer claims a transition that did not happen.
Index rows: this branch leaves the index arms as
mainhad them. A transacted Put passes no old image to the index sync and a transacted Delete reaches no index sync, so a key-changing Put leaves the old GSI row and a Delete removes none; on the service the GSI is correct within 4 s on both shapes. Filed as #568. With this branch a consumer sees a REMOVE record for a transacted delete while aQueryon the GSI still returns the item.The budget was measured on a single-node
cassandra:4.1with the default threshold and a string attribute carrying the size. The Delete arm's sequence-number order has no test, because a deleted row has no write time to compare against.Composition with #515
The stream capture is resolved before the PREPARE phase and the record is built in the commit loop, so the window between resolution and record write on the transaction path is the whole PREPARE phase (row reads, condition evaluation, marker LWTs, up to five create-race retries with 50 ms polls, the ledger update). #515 models a stream's shard as a sequence range of a physical partition that every write lands in regardless of the current stream; after it merges, a transacted write acked before a disable can land past the closed shard's ending number and inside no stream's range, and the count is not bounded by what #515's race probe measured, which exercised single-item writers only. Whoever merges second should measure transacted writers. Trial merges with #515 and #543 are clean.
Cassandra's
SequenceNumberis 23 characters where SQLite's is 21, and MongoDB numbers per shard from 1 where the other three use one counter; pre-existing in the single-item paths and unchanged here.