Skip to content

fix(postgres): order a PutItem that loses a create race after the winner #388

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yesyayen merged 11 commits into from
Oct 7, 2026
Merged

fix(postgres): order a PutItem that loses a create race after the winner #388

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
bc6a392
test: race PutItem calls that create the same item
yesyayen Oct 5, 2026
6f08081
fix(postgres): order a put that loses a create race after the winner
yesyayen Oct 5, 2026
d1d883f
test: check index and stream contents after racing puts
yesyayen Oct 5, 2026
3b73953
test: cover a put whose create race winner is deleted
yesyayen Oct 5, 2026
cebbba0
docs: say how a write to a missing item resolves a create race
yesyayen Oct 5, 2026
1d38037
test: run the create race gate tests on a range table too
yesyayen Oct 5, 2026
a31108e
test: say that four create race tests delete the winner
yesyayen Oct 5, 2026
f5515a3
test: expect the plain-table create races to fail on MongoDB for now
yesyayen Oct 5, 2026
18348d5
fix(postgres): make the create race give-up message true in every case
yesyayen Oct 7, 2026
faa624c
test: race a put against a create on an LSI and stream table
yesyayen Oct 7, 2026
f9e598a
test: count only an assertion as the expected MongoDB failure
yesyayen Oct 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions .github/workflows/integration.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,12 +297,14 @@ jobs:

# The control plane for vector indexes is not reachable over the wire while
# this backend declares no vector search capability, so its tests drive the
# storage layer directly against this job's PostgreSQL. They build their own
# throwaway databases; the connection string is the server, not a database.
# storage layer directly against this job's PostgreSQL. The PutItem create
# race tests hold a create open in an outside transaction, which also needs
# direct access. They build their own throwaway databases; the connection
# string is the server, not a database.
- name: Run PostgreSQL storage-level tests
env:
EXTENDDB_TEST_PG_CONNECTION_STRING: postgresql://postgres:devpass@127.0.0.1:5432
run: cargo test --release -p extenddb-storage-postgres --test vector_control_plane
run: cargo test --release -p extenddb-storage-postgres --test vector_control_plane --test put_create_race

# The daemonized server logs to syslog; dump it so server-side failures
# are diagnosable from the job log.
Expand Down
136 changes: 81 additions & 55 deletions crates/storage-postgres/src/data/put_item.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,25 +112,28 @@ impl PostgresEngine {
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;

let old: Option<(serde_json::Value,)> =
let mut old: Option<(serde_json::Value,)> =
bind_sk_fetch_optional!(&select_sql, pk_text.as_str(), &sk, &mut *tx)?;

if let Some((ref old_json,)) = old {
let old_item: Item = json_to_item(old_json.clone())?;
match check_condition(condition, &old_item, maps) {
Ok(()) => {}
Err(StorageError::ConditionFailed(_)) => {
return Err(StorageError::ConditionFailed(Some(old_item)));
let mut attempt: u32 = 0;
loop {
if let Some((ref old_json,)) = old {
let old_item: Item = json_to_item(old_json.clone())?;
match check_condition(condition, &old_item, maps) {
Ok(()) => {}
Err(StorageError::ConditionFailed(_)) => {
return Err(StorageError::ConditionFailed(Some(old_item)));
}
Err(e) => return Err(e),
}
Err(e) => return Err(e),
// Row exists, condition passed: update in place.
let update_sql = format!(
"UPDATE {ddb_table} SET item_data = $3 WHERE pk = $1 AND {sk_col} = $2"
);
bind_sk_execute!(&update_sql, pk_text.as_str(), &sk, &item_json, &mut *tx)?;
break;
}
// Row exists, condition passed — update in place.
let update_sql = format!(
"UPDATE {ddb_table} SET item_data = $3 WHERE pk = $1 AND {sk_col} = $2"
);
bind_sk_execute!(&update_sql, pk_text.as_str(), &sk, &item_json, &mut *tx)?;
} else {
// No existing item — condition checks against empty item
// No existing item: the condition checks against an empty item
let empty = std::collections::BTreeMap::new();
match check_condition(condition, &empty, maps) {
Ok(()) => {}
Expand All @@ -139,21 +142,25 @@ impl PostgresEngine {
}
Err(e) => return Err(e),
}
// Condition passed against empty — atomic insert, fail if someone beat us.
let insert_sql = format!(
"INSERT INTO {ddb_table} (pk, {sk_col}, item_data) VALUES ($1, $2, $3) \
ON CONFLICT (pk, {sk_col}) DO NOTHING"
);
let result =
bind_sk_execute!(&insert_sql, pk_text.as_str(), &sk, &item_json, &mut *tx)?;
if result.rows_affected() == 0 {
// Another transaction inserted between our SELECT and INSERT.
// Fetch the winner to return with ConditionFailed.
let winner: Option<(serde_json::Value,)> =
bind_sk_fetch_optional!(&select_sql, pk_text.as_str(), &sk, &mut *tx)?;
let winner_item = winner.map(|(v,)| json_to_item(v)).transpose()?;
return Err(StorageError::ConditionFailed(winner_item));
if result.rows_affected() == 1 {
break;
}
// Lost the create race. The winner has committed (the insert
// waited for it if it was in flight). The locking read returns
// its row, and this put overwrites it after checking its
// condition against it. If a delete committed since, the read
// returns no row and the insert is retried.
attempt += 1;
if attempt >= MAX_CREATE_RACE_ATTEMPTS {
return Err(create_race_exhausted(attempt));

@robinnsc robinnsc Oct 6, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The cap is checked after the fifth lost insert and before the re-read, so if the fifth winner is still present the put gives up rather than overwriting it. That matches UpdateItem exactly and the new boundary test pins it deliberately, so I don't think the behaviour is wrong, but the message "a concurrent writer repeatedly created and deleted the row" is not quite true in that case, since the last winner stayed. Could consider either one final re-read before the error so the message holds, or rewording the message to something like "lost the create race 5 times" so it stays accurate for both cases. Same applies to the hash-only branch at line 334.

@yesyayen yesyayen Oct 7, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, the message wasn't true when the last winner stays. Reworded to "each insert lost the create race to a concurrent writer". I kept the give-up point the same as UpdateItem.

}
old = bind_sk_fetch_optional!(&select_sql, pk_text.as_str(), &sk, &mut *tx)?;
}

// Sync GSI/LSI update within transaction (D-4).
Expand Down Expand Up @@ -269,30 +276,37 @@ impl PostgresEngine {
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;

let old: Option<(serde_json::Value,)> = sqlx::query_as(&select_sql)
.bind(pk_text.as_str())
.fetch_optional(&mut *tx)
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;

if let Some((ref old_json,)) = old {
let old_item: Item = json_to_item(old_json.clone())?;
match check_condition(condition, &old_item, maps) {
Ok(()) => {}
Err(StorageError::ConditionFailed(_)) => {
return Err(StorageError::ConditionFailed(Some(old_item)));
}
Err(e) => return Err(e),
}
// Row exists, condition passed — update in place.
let update_sql = format!("UPDATE {ddb_table} SET item_data = $2 WHERE pk = $1");
sqlx::query(&update_sql)
let fetch_old = async |tx: &mut sqlx::PgConnection| {
sqlx::query_as::<_, (serde_json::Value,)>(&select_sql)
.bind(pk_text.as_str())
.bind(&item_json)
.execute(&mut *tx)
.fetch_optional(tx)
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;
} else {
.map_err(|e| StorageError::Internal(e.to_string()))
};
let mut old: Option<(serde_json::Value,)> = fetch_old(&mut tx).await?;

let mut attempt: u32 = 0;
loop {
if let Some((ref old_json,)) = old {
let old_item: Item = json_to_item(old_json.clone())?;
match check_condition(condition, &old_item, maps) {
Ok(()) => {}
Err(StorageError::ConditionFailed(_)) => {
return Err(StorageError::ConditionFailed(Some(old_item)));
}
Err(e) => return Err(e),
}
// Row exists, condition passed: update in place.
let update_sql =
format!("UPDATE {ddb_table} SET item_data = $2 WHERE pk = $1");
sqlx::query(&update_sql)
.bind(pk_text.as_str())
.bind(&item_json)
.execute(&mut *tx)
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;
break;
}
let empty = std::collections::BTreeMap::new();
match check_condition(condition, &empty, maps) {
Ok(()) => {}
Expand All @@ -301,7 +315,6 @@ impl PostgresEngine {
}
Err(e) => return Err(e),
}
// Condition passed against empty — atomic insert, fail if someone beat us.
let insert_sql = format!(
"INSERT INTO {ddb_table} (pk, item_data) VALUES ($1, $2) \
ON CONFLICT (pk) DO NOTHING"
Expand All @@ -312,16 +325,15 @@ impl PostgresEngine {
.execute(&mut *tx)
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;
if result.rows_affected() == 0 {
// Another transaction inserted between our SELECT and INSERT.
let winner: Option<(serde_json::Value,)> = sqlx::query_as(&select_sql)
.bind(pk_text.as_str())
.fetch_optional(&mut *tx)
.await
.map_err(|e| StorageError::Internal(e.to_string()))?;
let winner_item = winner.map(|(v,)| json_to_item(v)).transpose()?;
return Err(StorageError::ConditionFailed(winner_item));
if result.rows_affected() == 1 {
break;
}
// Lost the create race: overwrite the winner, as above.
attempt += 1;
if attempt >= MAX_CREATE_RACE_ATTEMPTS {
return Err(create_race_exhausted(attempt));
}
old = fetch_old(&mut tx).await?;
}

// Sync GSI/LSI update within transaction (D-4).
Expand Down Expand Up @@ -466,3 +478,17 @@ impl PostgresEngine {
json_opt.map(json_to_item).transpose()
}
}

/// Bound on insert retries when a put to a missing item keeps losing the
/// create race to a winner that is deleted again before the re-read. Same
/// bound as `UpdateItem`.
const MAX_CREATE_RACE_ATTEMPTS: u32 = 5;

/// The error after `MAX_CREATE_RACE_ATTEMPTS` lost create races. Like the one
/// `UpdateItem` returns, it is an internal error (HTTP 500).
fn create_race_exhausted(attempt: u32) -> StorageError {
StorageError::Internal(format!(
"PutItem could not create the item after {attempt} attempts: each insert lost the \
create race to a concurrent writer"
))
}
Loading
Loading