mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-08-16 06:16:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(firehose): lost events if seq commits out of order

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-05-30 21:10:38 +03:00
committed by Tangled
parent 4e2525b245
commit e9dc57d6f4
48 changed files with 601 additions and 433 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.sqlx/query-09142b5a6b91fe4fd795fb6c85af5461e2621f6c3545b6d26e4618d813904b75.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "SELECT nextval('firehose_seq') as \"seq!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq!",
"type_info": "Int8"
}
],
"parameters": {
"Left": []
},
"nullable": [
null
]
},
"hash": "09142b5a6b91fe4fd795fb6c85af5461e2621f6c3545b6d26e4618d813904b75"
}
.sqlx/query-0b895d105363a60040323c5c47bc897905af3da980d0dc1b3b60bbdcdd90ea7a.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE repo_seq r\n SET seq = $1::bigint + (o.ord - 1)\n FROM unnest($2::bigint[]) WITH ORDINALITY AS o(id, ord)\n WHERE r.id = o.id",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Int8Array"
]
},
"nullable": []
},
"hash": "0b895d105363a60040323c5c47bc897905af3da980d0dc1b3b60bbdcdd90ea7a"
}
.sqlx/query-122d79e006c0e11214e3cf11cfb98bdff6fbf156893b482572532ebcd8979e61.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE FROM repo_seq\n WHERE did = $1\n AND id <> (SELECT id FROM repo_seq WHERE did = $1 ORDER BY id DESC LIMIT 1)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "122d79e006c0e11214e3cf11cfb98bdff6fbf156893b482572532ebcd8979e61"
}
.sqlx/query-a0ea3630adb9e411d94d88084df6b05d7fc3ba8c306146fc932b736fe342eceb.json → .sqlx/query-2c3c0740793f29b9bd9e0f81c2372d427dee091e20da8f71ae1e5f2769aa985d.json
Generated
+3 -3
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', false, 'deleted')\n RETURNING seq\n ",
"query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', false, 'deleted')\n RETURNING id\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"name": "id",
"type_info": "Int8"
}
],
@@ -18,5 +18,5 @@
false
]
},
"hash": "a0ea3630adb9e411d94d88084df6b05d7fc3ba8c306146fc932b736fe342eceb"
"hash": "2c3c0740793f29b9bd9e0f81c2372d427dee091e20da8f71ae1e5f2769aa985d"
}
.sqlx/query-3cc7247d2cd7e85a6f21d783dd956f8fecc6f63c994014bb4e91504d498b5fd5.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT pg_try_advisory_xact_lock($1) as \"locked!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "locked!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
null
]
},
"hash": "3cc7247d2cd7e85a6f21d783dd956f8fecc6f63c994014bb4e91504d498b5fd5"
}
.sqlx/query-4f3ef746cb5fc0916d3df5f92494c538497d3b9c1cfcc6e44c9b80f2193094f6.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO repo_seq (did, event_type, handle)\n VALUES ($1, 'identity', $2)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "4f3ef746cb5fc0916d3df5f92494c538497d3b9c1cfcc6e44c9b80f2193094f6"
}
.sqlx/query-50b2178cd174af1dc132a22048e52a6af492f4d8025ec512dd9c0c9ba62eb202.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO repo_seq (did, event_type, handle)\n VALUES ($1, 'identity', $2)\n RETURNING seq\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "50b2178cd174af1dc132a22048e52a6af492f4d8025ec512dd9c0c9ba62eb202"
}
.sqlx/query-714a1c9a73e2fed8182d4ab2fe4a2c33bc10a52e27ed95d3a6641477ae42395e.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id as \"id!\" FROM repo_seq WHERE seq IS NULL ORDER BY id LIMIT $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false
]
},
"hash": "714a1c9a73e2fed8182d4ab2fe4a2c33bc10a52e27ed95d3a6641477ae42395e"
}
.sqlx/query-4702a94384ce938d3f5384bc2106a48500f8749134e6bfd1c619a9151f5d3772.json → .sqlx/query-7cdf3b8136782a5b2e4aff91907a680566b4610e6cbf9d9396d31f9bbd99c3b8.json
Generated
+4 -4
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq = $1",
"query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"name": "seq!",
"type_info": "Int8"
},
{
@@ -90,7 +90,7 @@
]
},
"nullable": [
false,
true,
false,
false,
false,
@@ -108,5 +108,5 @@
true
]
},
"hash": "4702a94384ce938d3f5384bc2106a48500f8749134e6bfd1c619a9151f5d3772"
"hash": "7cdf3b8136782a5b2e4aff91907a680566b4610e6cbf9d9396d31f9bbd99c3b8"
}
.sqlx/query-a325862f897484bb1a8863fd7a1bc3fd17ab20b6e30498f9900daf80609c413d.json → .sqlx/query-800b93a38eef70bafdf0a1b221994f3c19ab562f84d18265376296a592b4c767.json
Generated
+4 -12
View File
@@ -1,14 +1,8 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, rev)\n VALUES ($1, 'commit', $2, $3::TEXT, $4, $5, $6, $7, $8)\n RETURNING seq\n ",
"query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, rev)\n VALUES ($1, 'commit', $2, $3::TEXT, $4, $5, $6, $7, $8)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"type_info": "Int8"
}
],
"columns": [],
"parameters": {
"Left": [
"Text",
@@ -21,9 +15,7 @@
"Text"
]
},
"nullable": [
false
]
"nullable": []
},
"hash": "a325862f897484bb1a8863fd7a1bc3fd17ab20b6e30498f9900daf80609c413d"
"hash": "800b93a38eef70bafdf0a1b221994f3c19ab562f84d18265376296a592b4c767"
}
.sqlx/query-97dd3cfe72b74b4dac686c030a07f6eb56dabb2df72815f3248760b4e3f7950e.json → .sqlx/query-82276a34f473ff8b9790b78915ae3f72873d4b748cdb47bd2b81edc2f7a6e5a8.json
Generated
+4 -12
View File
@@ -1,14 +1,8 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)\n RETURNING seq\n ",
"query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"type_info": "Int8"
}
],
"columns": [],
"parameters": {
"Left": [
"Text",
@@ -23,9 +17,7 @@
"Text"
]
},
"nullable": [
false
]
"nullable": []
},
"hash": "97dd3cfe72b74b4dac686c030a07f6eb56dabb2df72815f3248760b4e3f7950e"
"hash": "82276a34f473ff8b9790b78915ae3f72873d4b748cdb47bd2b81edc2f7a6e5a8"
}
.sqlx/query-b6d6548acb89d6384cd226f6ed0d66de27fde3af24b4a7a3fce7e098812e38a5.json → .sqlx/query-856e458c1c78f030545dc147f76d1b91fa4cc6e178eb51133c52358855ba3f02.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM repo_seq WHERE did = $1 AND seq != $2",
"query": "DELETE FROM repo_seq WHERE did = $1 AND id <> $2",
"describe": {
"columns": [],
"parameters": {
@@ -11,5 +11,5 @@
},
"nullable": []
},
"hash": "b6d6548acb89d6384cd226f6ed0d66de27fde3af24b4a7a3fce7e098812e38a5"
"hash": "856e458c1c78f030545dc147f76d1b91fa4cc6e178eb51133c52358855ba3f02"
}
.sqlx/query-623042c0b901e17ff37c5bcc1a4eefe56fc8b996ca8326e75796897d13c9e9ed.json → .sqlx/query-8ccd42b5b1b75f45e094889ee26af8517de28ca687efa81fd9c4fce598338fe9.json
Generated
+4 -12
View File
@@ -1,14 +1,8 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', $2, $3)\n RETURNING seq\n ",
"query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', $2, $3)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"type_info": "Int8"
}
],
"columns": [],
"parameters": {
"Left": [
"Text",
@@ -16,9 +10,7 @@
"Text"
]
},
"nullable": [
false
]
"nullable": []
},
"hash": "623042c0b901e17ff37c5bcc1a4eefe56fc8b996ca8326e75796897d13c9e9ed"
"hash": "8ccd42b5b1b75f45e094889ee26af8517de28ca687efa81fd9c4fce598338fe9"
}
.sqlx/query-c47746c256fd8c9d386551f3bd270e9a96fecdb713642d07d60fb4aeb02c8426.json → .sqlx/query-957bc81c63510ab502d2f8cea75dafcbdf2c8ebb448c86990d5e404ee5280438.json
Generated
+4 -4
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC",
"query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"name": "seq!",
"type_info": "Int8"
},
{
@@ -90,7 +90,7 @@
]
},
"nullable": [
false,
true,
false,
false,
false,
@@ -108,5 +108,5 @@
true
]
},
"hash": "c47746c256fd8c9d386551f3bd270e9a96fecdb713642d07d60fb4aeb02c8426"
"hash": "957bc81c63510ab502d2f8cea75dafcbdf2c8ebb448c86990d5e404ee5280438"
}
.sqlx/query-5996774d097a484f81353d49e4a99b07c30a003377c924add576bfb48ddde29b.json → .sqlx/query-9ba7869f29286cfd6b811ffa1e26bda1d637b98b44be66cdc265b1af4cec2d4f.json
Generated
+4 -12
View File
@@ -1,14 +1,8 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, rev, block_cids, block_data)\n VALUES ($1, 'sync', $2, $3, $4, $5)\n RETURNING seq\n ",
"query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, rev, block_cids, block_data)\n VALUES ($1, 'sync', $2, $3, $4, $5)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"type_info": "Int8"
}
],
"columns": [],
"parameters": {
"Left": [
"Text",
@@ -18,9 +12,7 @@
"ByteaArray"
]
},
"nullable": [
false
]
"nullable": []
},
"hash": "5996774d097a484f81353d49e4a99b07c30a003377c924add576bfb48ddde29b"
"hash": "9ba7869f29286cfd6b811ffa1e26bda1d637b98b44be66cdc265b1af4cec2d4f"
}
.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT pg_advisory_xact_lock($1)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "pg_advisory_xact_lock",
"type_info": "Void"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
null
]
},
"hash": "a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247"
}
.sqlx/query-bd0407818c8bdae1f1024777951b08cb9c386fe751722e1caf6ad243e271697a.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT setval('firehose_seq', $1)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "setval",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
null
]
},
"hash": "bd0407818c8bdae1f1024777951b08cb9c386fe751722e1caf6ad243e271697a"
}
.sqlx/query-d4eacb9bca838dd18dcfd8ca39b83ef50968bfaaabc213536c049cc222ffe9c8.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "NOTIFY repo_updates",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "d4eacb9bca838dd18dcfd8ca39b83ef50968bfaaabc213536c049cc222ffe9c8"
}
.sqlx/query-89c8ce80d8c52b4668c34f63e48b51cacc7e8ff300958a192378f1d72de04f9a.json → .sqlx/query-da2770e44c0ec21756ce4b6b0c57abff4f38cd8c680c5ced7d3b35f3c1f58da7.json
Generated
+4 -4
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2",
"query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"name": "seq!",
"type_info": "Int8"
},
{
@@ -91,7 +91,7 @@
]
},
"nullable": [
false,
true,
false,
false,
false,
@@ -109,5 +109,5 @@
true
]
},
"hash": "89c8ce80d8c52b4668c34f63e48b51cacc7e8ff300958a192378f1d72de04f9a"
"hash": "da2770e44c0ec21756ce4b6b0c57abff4f38cd8c680c5ced7d3b35f3c1f58da7"
}
.sqlx/query-0349b2af4d21105029ecc2bf81ad7be9987ec17ac088730596849d7a45558e57.json → .sqlx/query-dfe011b2d2bafbbbbea508a4977fc34bb1be4cf2612296df23763e6e5841c33e.json
Generated
+4 -4
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2",
"query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1 AND seq < $2\n ORDER BY seq ASC",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"name": "seq!",
"type_info": "Int8"
},
{
@@ -91,7 +91,7 @@
]
},
"nullable": [
false,
true,
false,
false,
false,
@@ -109,5 +109,5 @@
true
]
},
"hash": "0349b2af4d21105029ecc2bf81ad7be9987ec17ac088730596849d7a45558e57"
"hash": "dfe011b2d2bafbbbbea508a4977fc34bb1be4cf2612296df23763e6e5841c33e"
}
.sqlx/query-ad784aff9bb5c1f19f90a3b1c3b3694a3e1aa889b0fe3e55cfebb7af8c620256.json → .sqlx/query-f16d41b41a5388997e6ed408da55056173340ed12b60b1a956de87f73f9d0824.json
Generated
+4 -4
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1 AND seq < $2\n ORDER BY seq ASC",
"query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"name": "seq!",
"type_info": "Int8"
},
{
@@ -91,7 +91,7 @@
]
},
"nullable": [
false,
true,
false,
false,
false,
@@ -109,5 +109,5 @@
true
]
},
"hash": "ad784aff9bb5c1f19f90a3b1c3b3694a3e1aa889b0fe3e55cfebb7af8c620256"
"hash": "f16d41b41a5388997e6ed408da55056173340ed12b60b1a956de87f73f9d0824"
}
crates/tranquil-api/src/identity/plc/submit.rs
+2 -9
View File
@@ -129,20 +129,13 @@ pub async fn submit_plc_operation(
.await
.map_err(ApiError::from)?;
match state
if let Err(e) = state
.repos
.repo
.insert_identity_event(did, Some(&user.handle))
.await
{
Ok(seq) => {
if let Err(e) = state.repos.repo.notify_update(seq).await {
warn!("Failed to notify identity event: {:?}", e);
}
}
Err(e) => {
warn!("Failed to sequence identity event: {:?}", e);
}
warn!("Failed to sequence identity event: {:?}", e);
}
let _ = state
.cache
crates/tranquil-api/src/repo/import.rs
+1 -2
View File
@@ -417,7 +417,6 @@ async fn sequence_import_event(
rev: None,
};
let seq = state.repos.repo.insert_commit_event(&data).await?;
state.repos.repo.notify_update(seq).await?;
state.repos.repo.insert_commit_event(&data).await?;
Ok(())
}
crates/tranquil-api/src/server/account_status.rs
+12 -17
View File
@@ -665,27 +665,22 @@ pub async fn delete_account(
error!("DB error deleting account: {:?}", e);
ApiError::InternalError(None)
})?;
let account_seq = tranquil_pds::repo_ops::sequence_account_event(
if let Err(e) = tranquil_pds::repo_ops::sequence_account_event(
&state,
did,
tranquil_db_traits::AccountStatus::Deleted,
)
.await;
match account_seq {
Ok(seq) => {
if let Err(e) = state.repos.repo.delete_sequences_except(did, seq).await {
warn!(
"Failed to cleanup sequences for deleted account {}: {}",
did, e
);
}
}
Err(e) => {
warn!(
"Failed to sequence account deletion event for {}: {}",
did, e
);
}
.await
{
warn!(
"Failed to sequence account deletion event for {}: {}",
did, e
);
} else if let Err(e) = state.repos.repo.purge_did_events_keeping_latest(did).await {
warn!(
"Failed to cleanup sequences for deleted account {}: {}",
did, e
);
}
let _ = state
.cache
crates/tranquil-db-traits/src/repo.rs
+15 -18
View File
@@ -319,7 +319,6 @@ pub struct ApplyCommitInput {
#[derive(Debug, Clone)]
pub struct ApplyCommitResult {
pub seq: i64,
pub is_account_active: bool,
}
@@ -445,19 +444,15 @@ pub trait RepoRepository: Send + Sync {
async fn count_user_blocks(&self, user_id: Uuid) -> Result<i64, DbError>;
async fn insert_commit_event(&self, data: &CommitEventData) -> Result<SequenceNumber, DbError>;
async fn insert_commit_event(&self, data: &CommitEventData) -> Result<(), DbError>;
async fn insert_identity_event(
&self,
did: &Did,
handle: Option<&Handle>,
) -> Result<SequenceNumber, DbError>;
) -> Result<(), DbError>;
async fn insert_account_event(
&self,
did: &Did,
status: AccountStatus,
) -> Result<SequenceNumber, DbError>;
async fn insert_account_event(&self, did: &Did, status: AccountStatus) -> Result<(), DbError>;
async fn insert_sync_event(
&self,
@@ -465,7 +460,7 @@ pub trait RepoRepository: Send + Sync {
commit_cid: &CidLink,
rev: Option<&str>,
commit_bytes: &[u8],
) -> Result<SequenceNumber, DbError>;
) -> Result<(), DbError>;
async fn insert_genesis_commit_event(
&self,
@@ -475,14 +470,18 @@ pub trait RepoRepository: Send + Sync {
rev: &str,
commit_bytes: &[u8],
mst_root_bytes: &[u8],
) -> Result<SequenceNumber, DbError>;
async fn delete_sequences_except(
&self,
did: &Did,
keep_seq: SequenceNumber,
) -> Result<(), DbError>;
async fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError>;
async fn assign_pending_sequences(&self) -> Result<u64, DbError> {
Ok(0)
}
async fn flush_pending_sequences(&self) -> Result<(), DbError> {
Ok(())
}
async fn prune_events_older_than(&self, cutoff: DateTime<Utc>) -> Result<PruneCount, DbError>;
async fn get_max_seq(&self) -> Result<SequenceNumber, DbError>;
@@ -526,8 +525,6 @@ pub trait RepoRepository: Send + Sync {
async fn get_repo_root_cid_by_user_id(&self, user_id: Uuid)
-> Result<Option<CidLink>, DbError>;
async fn notify_update(&self, seq: SequenceNumber) -> Result<(), DbError>;
async fn import_repo_data(
&self,
user_id: Uuid,
@@ -563,5 +560,5 @@ pub trait RepoEventNotifier: Send + Sync {
#[async_trait]
pub trait RepoEventReceiver: Send {
async fn recv(&mut self) -> Option<i64>;
async fn recv(&mut self) -> Option<()>;
}
crates/tranquil-db-traits/src/user.rs
+1 -1
View File
@@ -551,7 +551,7 @@ pub trait UserRepository: Send + Sync {
limit: i64,
) -> Result<Vec<ScheduledDeletionAccount>, DbError>;
async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<i64, DbError>;
async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError>;
async fn create_password_account(
&self,
crates/tranquil-db/src/postgres/event_notifier.rs
+2 -2
View File
@@ -35,9 +35,9 @@ pub struct PostgresRepoEventReceiver {
#[async_trait]
impl RepoEventReceiver for PostgresRepoEventReceiver {
async fn recv(&mut self) -> Option<i64> {
async fn recv(&mut self) -> Option<()> {
match self.listener.recv().await {
Ok(notification) => notification.payload().parse().ok(),
Ok(_) => Some(()),
Err(_) => None,
}
}
crates/tranquil-db/src/postgres/repo.rs
+145 -88
View File
@@ -111,6 +111,64 @@ fn map_sequenced_row(r: SequencedEventRow) -> Result<SequencedEvent, DbError> {
})
}
const SEQUENCER_LOCK_KEY: i64 = 0x0074_7261_6e73_6571;
const SEQUENCER_BATCH_SIZE: i64 = 1000;
async fn notify_repo_pending(pool: &PgPool) {
if let Err(e) = sqlx::query!("NOTIFY repo_updates").execute(pool).await {
tracing::warn!(error = %e, "failed to notify repo_updates listeners");
}
}
async fn assign_one_batch(
mut tx: sqlx::Transaction<'_, sqlx::Postgres>,
pool: &PgPool,
) -> Result<i64, DbError> {
let pending_ids: Vec<i64> = sqlx::query_scalar!(
r#"SELECT id as "id!" FROM repo_seq WHERE seq IS NULL ORDER BY id LIMIT $1"#,
SEQUENCER_BATCH_SIZE
)
.fetch_all(&mut *tx)
.await
.map_err(map_sqlx_error)?;
let count = pending_ids.len() as i64;
if count == 0 {
tx.commit().await.map_err(map_sqlx_error)?;
return Ok(0);
}
let start = sqlx::query_scalar!(r#"SELECT nextval('firehose_seq') as "seq!""#)
.fetch_one(&mut *tx)
.await
.map_err(map_sqlx_error)?;
if count > 1 {
sqlx::query!("SELECT setval('firehose_seq', $1)", start + count - 1)
.fetch_one(&mut *tx)
.await
.map_err(map_sqlx_error)?;
}
let assigned = sqlx::query!(
r#"UPDATE repo_seq r
SET seq = $1::bigint + (o.ord - 1)
FROM unnest($2::bigint[]) WITH ORDINALITY AS o(id, ord)
WHERE r.id = o.id"#,
start,
&pending_ids
)
.execute(&mut *tx)
.await
.map_err(map_sqlx_error)?
.rows_affected();
tx.commit().await.map_err(map_sqlx_error)?;
if assigned > 0 {
notify_repo_pending(pool).await;
}
Ok(count)
}
pub struct PostgresRepoRepository {
pool: PgPool,
}
@@ -716,13 +774,12 @@ impl RepoRepository for PostgresRepoRepository {
Ok(rows.into_iter().map(|(cid,)| cid).collect())
}
async fn insert_commit_event(&self, data: &CommitEventData) -> Result<SequenceNumber, DbError> {
async fn insert_commit_event(&self, data: &CommitEventData) -> Result<(), DbError> {
let (block_cids, block_data) = inline_to_paired_blocks(data.blocks.as_deref());
let seq = sqlx::query_scalar!(
sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
RETURNING seq
"#,
data.did.as_str(),
data.event_type.as_str(),
@@ -735,67 +792,54 @@ impl RepoRepository for PostgresRepoRepository {
data.prev_data_cid.as_ref().map(|c| c.as_str()),
data.rev
)
.fetch_one(&self.pool)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(seq.into())
notify_repo_pending(&self.pool).await;
Ok(())
}
async fn insert_identity_event(
&self,
did: &Did,
handle: Option<&Handle>,
) -> Result<SequenceNumber, DbError> {
) -> Result<(), DbError> {
let handle_str = handle.map(|h| h.as_str());
let seq = sqlx::query_scalar!(
sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, handle)
VALUES ($1, 'identity', $2)
RETURNING seq
"#,
did.as_str(),
handle_str
)
.fetch_one(&self.pool)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq))
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(seq.into())
notify_repo_pending(&self.pool).await;
Ok(())
}
async fn insert_account_event(
&self,
did: &Did,
status: AccountStatus,
) -> Result<SequenceNumber, DbError> {
async fn insert_account_event(&self, did: &Did, status: AccountStatus) -> Result<(), DbError> {
let active = status.is_active();
let status_str = status.for_firehose();
let seq = sqlx::query_scalar!(
sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, active, status)
VALUES ($1, 'account', $2, $3)
RETURNING seq
"#,
did.as_str(),
active,
status_str
)
.fetch_one(&self.pool)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq))
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(seq.into())
notify_repo_pending(&self.pool).await;
Ok(())
}
async fn insert_sync_event(
@@ -804,18 +848,17 @@ impl RepoRepository for PostgresRepoRepository {
commit_cid: &CidLink,
rev: Option<&str>,
commit_bytes: &[u8],
) -> Result<SequenceNumber, DbError> {
) -> Result<(), DbError> {
let cid_bytes = commit_cid
.to_cid()
.map(|c| c.to_bytes())
.unwrap_or_default();
let block_cids: Vec<Vec<u8>> = vec![cid_bytes];
let block_data: Vec<Vec<u8>> = vec![commit_bytes.to_vec()];
let seq = sqlx::query_scalar!(
sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, commit_cid, rev, block_cids, block_data)
VALUES ($1, 'sync', $2, $3, $4, $5)
RETURNING seq
"#,
did.as_str(),
commit_cid.as_str(),
@@ -823,16 +866,12 @@ impl RepoRepository for PostgresRepoRepository {
&block_cids as &[Vec<u8>],
&block_data as &[Vec<u8>]
)
.fetch_one(&self.pool)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq))
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(seq.into())
notify_repo_pending(&self.pool).await;
Ok(())
}
async fn insert_genesis_commit_event(
@@ -843,7 +882,7 @@ impl RepoRepository for PostgresRepoRepository {
rev: &str,
commit_bytes: &[u8],
mst_root_bytes: &[u8],
) -> Result<SequenceNumber, DbError> {
) -> Result<(), DbError> {
let ops = serde_json::json!([]);
let blobs: Vec<String> = vec![];
let commit_cid_bytes = commit_cid
@@ -858,11 +897,10 @@ impl RepoRepository for PostgresRepoRepository {
let block_data: Vec<Vec<u8>> = vec![commit_bytes.to_vec(), mst_root_bytes.to_vec()];
let prev_cid: Option<&str> = None;
let seq = sqlx::query_scalar!(
sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, rev)
VALUES ($1, 'commit', $2, $3::TEXT, $4, $5, $6, $7, $8)
RETURNING seq
"#,
did.as_str(),
commit_cid.as_str(),
@@ -873,27 +911,22 @@ impl RepoRepository for PostgresRepoRepository {
&block_data as &[Vec<u8>],
rev
)
.fetch_one(&self.pool)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq))
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(seq.into())
notify_repo_pending(&self.pool).await;
Ok(())
}
async fn delete_sequences_except(
&self,
did: &Did,
keep_seq: SequenceNumber,
) -> Result<(), DbError> {
async fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError> {
sqlx::query!(
"DELETE FROM repo_seq WHERE did = $1 AND seq != $2",
did.as_str(),
keep_seq.as_i64()
r#"
DELETE FROM repo_seq
WHERE did = $1
AND id <> (SELECT id FROM repo_seq WHERE did = $1 ORDER BY id DESC LIMIT 1)
"#,
did.as_str()
)
.execute(&self.pool)
.await
@@ -902,6 +935,42 @@ impl RepoRepository for PostgresRepoRepository {
Ok(())
}
async fn assign_pending_sequences(&self) -> Result<u64, DbError> {
let mut total: u64 = 0;
loop {
let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
let acquired = sqlx::query_scalar!(
r#"SELECT pg_try_advisory_xact_lock($1) as "locked!""#,
SEQUENCER_LOCK_KEY
)
.fetch_one(&mut *tx)
.await
.map_err(map_sqlx_error)?;
if !acquired {
return Ok(total);
}
let count = assign_one_batch(tx, &self.pool).await?;
total += count as u64;
if count < SEQUENCER_BATCH_SIZE {
return Ok(total);
}
}
}
async fn flush_pending_sequences(&self) -> Result<(), DbError> {
loop {
let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!("SELECT pg_advisory_xact_lock($1)", SEQUENCER_LOCK_KEY)
.execute(&mut *tx)
.await
.map_err(map_sqlx_error)?;
let count = assign_one_batch(tx, &self.pool).await?;
if count < SEQUENCER_BATCH_SIZE {
return Ok(());
}
}
}
async fn prune_events_older_than(&self, cutoff: DateTime<Utc>) -> Result<PruneCount, DbError> {
let result = sqlx::query!("DELETE FROM repo_seq WHERE created_at < $1", cutoff)
.execute(&self.pool)
@@ -965,7 +1034,7 @@ impl RepoRepository for PostgresRepoRepository {
Some(lim) => {
let rows = sqlx::query_as!(
SequencedEventRow,
r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev
FROM repo_seq
WHERE seq > $1
@@ -982,7 +1051,7 @@ impl RepoRepository for PostgresRepoRepository {
None => {
let rows = sqlx::query_as!(
SequencedEventRow,
r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev
FROM repo_seq
WHERE seq > $1
@@ -1004,7 +1073,7 @@ impl RepoRepository for PostgresRepoRepository {
) -> Result<Vec<SequencedEvent>, DbError> {
let rows = sqlx::query_as!(
SequencedEventRow,
r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev
FROM repo_seq
WHERE seq > $1 AND seq < $2
@@ -1024,7 +1093,7 @@ impl RepoRepository for PostgresRepoRepository {
) -> Result<Option<SequencedEvent>, DbError> {
let row = sqlx::query_as!(
SequencedEventRow,
r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev
FROM repo_seq
WHERE seq = $1"#,
@@ -1043,7 +1112,7 @@ impl RepoRepository for PostgresRepoRepository {
) -> Result<Vec<SequencedEvent>, DbError> {
let rows = sqlx::query_as!(
SequencedEventRow,
r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid,
ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev
FROM repo_seq
WHERE seq > $1
@@ -1104,14 +1173,6 @@ impl RepoRepository for PostgresRepoRepository {
Ok(cid.map(CidLink::from))
}
async fn notify_update(&self, seq: SequenceNumber) -> Result<(), DbError> {
sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq.as_i64()))
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn import_repo_data(
&self,
user_id: Uuid,
@@ -1402,28 +1463,27 @@ impl RepoRepository for PostgresRepoRepository {
let event = input.commit_event;
let (event_block_cids, event_block_data) = inline_into_paired_blocks(event.blocks);
let seq: i64 = sqlx::query_scalar(
sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
RETURNING seq
"#,
event.did.as_str(),
event.event_type.as_str(),
event.commit_cid.as_ref().map(|c| c.as_str()),
event.prev_cid.as_ref().map(|c| c.as_str()),
event.ops,
event.blobs.as_deref(),
&event_block_cids as &[Vec<u8>],
&event_block_data as &[Vec<u8>],
event.prev_data_cid.as_ref().map(|c| c.as_str()),
event.rev
)
.bind(event.did.as_str())
.bind(event.event_type.as_str())
.bind(event.commit_cid.as_ref().map(|c| c.as_str()))
.bind(event.prev_cid.as_ref().map(|c| c.as_str()))
.bind(&event.ops)
.bind(&event.blobs)
.bind(&event_block_cids)
.bind(&event_block_data)
.bind(event.prev_data_cid.as_ref().map(|c| c.as_str()))
.bind(&event.rev)
.fetch_one(&mut *tx)
.execute(&mut *tx)
.await
.map_err(|e| ApplyCommitError::Database(e.to_string()))?;
sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq))
sqlx::query!("NOTIFY repo_updates")
.execute(&mut *tx)
.await
.map_err(|e| ApplyCommitError::Database(e.to_string()))?;
@@ -1432,10 +1492,7 @@ impl RepoRepository for PostgresRepoRepository {
.await
.map_err(|e| ApplyCommitError::Database(e.to_string()))?;
Ok(tranquil_db_traits::ApplyCommitResult {
seq,
is_account_active,
})
Ok(tranquil_db_traits::ApplyCommitResult { is_account_active })
}
async fn get_users_without_blocks(&self) -> Result<Vec<UserWithoutBlocks>, DbError> {
crates/tranquil-db/src/postgres/user.rs
+7 -7
View File
@@ -2291,7 +2291,7 @@ impl UserRepository for PostgresUserRepository {
.collect())
}
async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<i64, DbError> {
async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> {
let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id)
@@ -2370,11 +2370,11 @@ impl UserRepository for PostgresUserRepository {
.await
.map_err(map_sqlx_error)?;
let account_seq: i64 = sqlx::query_scalar!(
let event_id: i64 = sqlx::query_scalar!(
r#"
INSERT INTO repo_seq (did, event_type, active, status)
VALUES ($1, 'account', false, 'deleted')
RETURNING seq
RETURNING id
"#,
did.as_str()
)
@@ -2383,9 +2383,9 @@ impl UserRepository for PostgresUserRepository {
.map_err(map_sqlx_error)?;
sqlx::query!(
"DELETE FROM repo_seq WHERE did = $1 AND seq != $2",
"DELETE FROM repo_seq WHERE did = $1 AND id <> $2",
did.as_str(),
account_seq
event_id
)
.execute(&mut *tx)
.await
@@ -2393,12 +2393,12 @@ impl UserRepository for PostgresUserRepository {
tx.commit().await.map_err(map_sqlx_error)?;
sqlx::query(&format!("NOTIFY repo_updates, '{}'", account_seq))
sqlx::query!("NOTIFY repo_updates")
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(account_seq)
Ok(())
}
async fn create_password_account(
crates/tranquil-pds/src/repo_ops.rs
+4 -5
View File
@@ -19,7 +19,6 @@ use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::OwnedMutexGuard;
use tracing::{error, warn};
use tranquil_db_traits::SequenceNumber;
use uuid::Uuid;
#[derive(Debug)]
@@ -776,7 +775,7 @@ pub async fn sequence_identity_event(
state: &AppState,
did: &Did,
handle: Option<&Handle>,
) -> Result<SequenceNumber, CommitError> {
) -> Result<(), CommitError> {
state
.repos
.repo
@@ -788,7 +787,7 @@ pub async fn sequence_account_event(
state: &AppState,
did: &Did,
status: tranquil_db_traits::AccountStatus,
) -> Result<SequenceNumber, CommitError> {
) -> Result<(), CommitError> {
state
.repos
.repo
@@ -801,7 +800,7 @@ pub async fn sequence_sync_event(
did: &Did,
commit_cid: &str,
rev: Option<&str>,
) -> Result<SequenceNumber, CommitError> {
) -> Result<(), CommitError> {
let cid_link: crate::types::CidLink = commit_cid
.parse()
.map_err(|_| CommitError::InvalidCid(commit_cid.to_string()))?;
@@ -829,7 +828,7 @@ pub async fn sequence_genesis_commit(
commit_cid: &Cid,
mst_root_cid: &Cid,
rev: &str,
) -> Result<SequenceNumber, CommitError> {
) -> Result<(), CommitError> {
let commit_cid_link = crate::types::CidLink::from(commit_cid);
let mst_root_cid_link = crate::types::CidLink::from(mst_root_cid);
let commit_bytes = state
crates/tranquil-pds/src/scheduled.rs
+1 -1
View File
@@ -711,7 +711,7 @@ async fn delete_account_data(
warn!(storage_key = %key, error = %e, "Failed to delete blob from storage (continuing anyway)");
});
let _account_seq = user_repo
user_repo
.delete_account_with_firehose(user_id, did)
.await
.context("Failed to delete account")?;
crates/tranquil-pds/tests/common/mod.rs
+34
View File
@@ -921,6 +921,40 @@ pub async fn get_test_block_store() -> &'static tranquil_pds::repo::AnyBlockStor
.expect("TEST_BLOCK_STORE not initialized")
}
#[allow(dead_code)]
pub async fn flushed_max_seq(
repos: &tranquil_db::PostgresRepositories,
) -> tranquil_db_traits::SequenceNumber {
repos
.repo
.flush_pending_sequences()
.await
.expect("flush_pending_sequences");
repos.repo.get_max_seq().await.expect("get_max_seq")
}
#[allow(dead_code)]
pub async fn sequenced_event_for_did(
repos: &tranquil_db::PostgresRepositories,
baseline: tranquil_db_traits::SequenceNumber,
did: &tranquil_types::Did,
) -> tranquil_db_traits::SequencedEvent {
repos
.repo
.flush_pending_sequences()
.await
.expect("flush_pending_sequences");
repos
.repo
.get_events_since_seq(baseline, None)
.await
.expect("get_events_since_seq")
.into_iter()
.filter(|event| &event.did == did)
.last()
.unwrap_or_else(|| panic!("event for did {did} not found after flush"))
}
fn extract_verification_code(body_text: &str) -> String {
let lines: Vec<&str> = body_text.lines().collect();
lines
crates/tranquil-pds/tests/firehose_inline_blocks.rs
+16 -14
View File
@@ -66,6 +66,11 @@ async fn commit_events_carry_inline_blocks() {
let repos = get_test_repos().await;
let typed_did = tranquil_types::Did::new(did.clone()).unwrap();
repos
.repo
.flush_pending_sequences()
.await
.expect("flush_pending_sequences");
let events = repos
.repo
.get_events_since_seq(SequenceNumber::ZERO, None)
@@ -142,18 +147,14 @@ async fn sync_event_carries_inline_commit_block() {
let cid_link: CidLink = (&commit_cid).into();
let rev = "3kabcdefghij2";
let seq = repos
let baseline = repos.repo.get_max_seq().await.expect("get_max_seq");
repos
.repo
.insert_sync_event(&did, &cid_link, Some(rev), &commit_bytes)
.await
.expect("insert_sync_event");
let event = repos
.repo
.get_event_by_seq(seq)
.await
.expect("get_event_by_seq")
.expect("event present");
let event = sequenced_event_for_did(repos, baseline, &did).await;
assert_eq!(event.event_type, RepoEventType::Sync);
let blocks = event
@@ -193,7 +194,8 @@ async fn genesis_commit_event_carries_inline_blocks() {
let mst_link: CidLink = (&mst_root_cid).into();
let rev = "3kabcdefghij3";
let seq = repos
let baseline = repos.repo.get_max_seq().await.expect("get_max_seq");
repos
.repo
.insert_genesis_commit_event(
&did,
@@ -206,12 +208,7 @@ async fn genesis_commit_event_carries_inline_blocks() {
.await
.expect("insert_genesis_commit_event");
let event = repos
.repo
.get_event_by_seq(seq)
.await
.expect("get_event_by_seq")
.expect("event present");
let event = sequenced_event_for_did(repos, baseline, &did).await;
assert_eq!(event.event_type, RepoEventType::Commit);
let blocks = event
@@ -329,6 +326,11 @@ async fn import_event_carries_inline_commit_block() {
let repos = get_test_repos().await;
let typed_did = tranquil_types::Did::new(did.clone()).unwrap();
repos
.repo
.flush_pending_sequences()
.await
.expect("flush_pending_sequences");
let events = repos
.repo
.get_events_since_seq(SequenceNumber::ZERO, None)
crates/tranquil-pds/tests/firehose_validation.rs
+1 -1
View File
@@ -801,7 +801,7 @@ async fn test_firehose_outdated_cursor_info() {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let repos = get_test_repos().await;
let max_seq = repos.repo.get_max_seq().await.unwrap().as_i64();
let max_seq = flushed_max_seq(repos).await.as_i64();
let outdated_cursor = (max_seq - 100).max(1);
let url = format!(
"ws://127.0.0.1:{}/xrpc/com.atproto.sync.subscribeRepos?cursor={}",
crates/tranquil-pds/tests/mst_firehose_e2e.rs
+5 -5
View File
@@ -223,7 +223,7 @@ async fn websocket_firehose_frames_pass_inductive_forward_and_inverse() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(Duration::from_millis(100)).await;
@@ -298,7 +298,7 @@ async fn websocket_firehose_car_root_matches_commit_cid() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(Duration::from_millis(100)).await;
@@ -340,7 +340,7 @@ async fn websocket_firehose_resumption_from_cursor_yields_valid_frames() {
create_record(&client, &token, &did, &rkey_for(i), "pre").await;
}
let resume_cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let resume_cursor = flushed_max_seq(repos).await.as_i64();
for i in 5..12 {
create_record(&client, &token, &did, &rkey_for(i), "post").await;
@@ -371,7 +371,7 @@ async fn websocket_firehose_ops_include_prev_field_for_update_delete() {
let client = client();
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(Duration::from_millis(100)).await;
@@ -417,7 +417,7 @@ async fn websocket_firehose_rebuild_new_mst_from_car_matches_commit_data() {
let client = client();
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(Duration::from_millis(100)).await;
crates/tranquil-pds/tests/mst_inductive_firehose.rs
+5
View File
@@ -247,6 +247,11 @@ fn rkey_for(prefix: &str, i: usize) -> String {
async fn our_commit_events(did: &str) -> Vec<SequencedEvent> {
let repos = get_test_repos().await;
let typed_did = Did::new(did.to_string()).unwrap();
repos
.repo
.flush_pending_sequences()
.await
.expect("flush_pending_sequences");
let events = repos
.repo
.get_events_since_seq(SequenceNumber::ZERO, None)
crates/tranquil-pds/tests/mst_integrity.rs
+7 -2
View File
@@ -216,6 +216,11 @@ async fn imported_repo_emits_commit_event_with_valid_car() {
let repos = get_test_repos().await;
let typed_did = Did::new(did.clone()).unwrap();
repos
.repo
.flush_pending_sequences()
.await
.expect("flush_pending_sequences");
let events = repos
.repo
.get_events_since_seq(SequenceNumber::ZERO, None)
@@ -253,7 +258,7 @@ async fn firehose_commit_block_bytes_roundtrip_to_same_cid() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(Duration::from_millis(100)).await;
@@ -284,7 +289,7 @@ async fn firehose_commit_car_contains_new_record_bytes_for_every_create() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(Duration::from_millis(100)).await;
crates/tranquil-pds/tests/mst_property_fuzz.rs
+5
View File
@@ -267,6 +267,11 @@ async fn fuzz_run_with_seed(seed: u64, steps: usize) -> Vec<String> {
let repos = get_test_repos().await;
let typed_did = Did::new(did.clone()).unwrap();
repos
.repo
.flush_pending_sequences()
.await
.expect("flush_pending_sequences");
let events = repos
.repo
.get_events_since_seq(SequenceNumber::ZERO, None)
crates/tranquil-pds/tests/repo_lifecycle.rs
+8 -8
View File
@@ -59,7 +59,7 @@ async fn test_create_record_cid_matches_firehose() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -134,7 +134,7 @@ async fn test_update_record_prev_matches_old_cid() {
let v1_cid = Cid::from_str(v1_cid_str).unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -203,7 +203,7 @@ async fn test_delete_record_prev_set_cid_none() {
let rkey = parts[parts.len() - 1];
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -246,7 +246,7 @@ async fn test_five_record_commit_chain_integrity() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -315,7 +315,7 @@ async fn test_apply_writes_single_commit_multiple_ops() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -396,7 +396,7 @@ async fn test_firehose_commit_signature_verification() {
};
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -444,7 +444,7 @@ async fn test_cursor_backfill_completeness() {
let (token, did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let baseline_seq = repos.repo.get_max_seq().await.unwrap().as_i64();
let baseline_seq = flushed_max_seq(repos).await.as_i64();
let mut expected_cids: Vec<String> = Vec::with_capacity(5);
let texts = [
@@ -496,7 +496,7 @@ async fn test_multi_account_seq_interleaving() {
let (bob_token, bob_did) = create_account_and_login(&client).await;
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let cursor = flushed_max_seq(repos).await.as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
crates/tranquil-pds/tests/store_parity.rs
+6 -4
View File
@@ -1628,10 +1628,12 @@ async fn parity_prune_events_older_than() {
rev: Some("rev0".to_string()),
};
let pg_seq = f.pg.repo.insert_commit_event(&event).await.unwrap();
let store_seq = f.store.repo.insert_commit_event(&event).await.unwrap();
assert!(pg_seq.as_i64() > 0);
assert!(store_seq.as_i64() > 0);
let baseline = f.pg.repo.get_max_seq().await.unwrap();
f.pg.repo.insert_commit_event(&event).await.unwrap();
f.store.repo.insert_commit_event(&event).await.unwrap();
let pg_seq = common::sequenced_event_for_did(&f.pg, baseline, &did)
.await
.seq;
let past_cutoff = chrono::Utc::now() - chrono::Duration::hours(24);
let pg_pruned_past =
crates/tranquil-store/src/eventlog/notifier.rs
+2 -2
View File
@@ -30,7 +30,7 @@ struct EventLogEventReceiver<S: StorageIO> {
#[async_trait]
impl<S: StorageIO + 'static> RepoEventReceiver for EventLogEventReceiver<S> {
async fn recv(&mut self) -> Option<i64> {
self.subscriber.next().await.map(|event| event.seq.as_i64())
async fn recv(&mut self) -> Option<()> {
self.subscriber.next().await.map(|_| ())
}
}
crates/tranquil-store/src/metastore/client.rs
+13 -32
View File
@@ -474,21 +474,21 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
recv(rx).await
}
async fn insert_commit_event(&self, data: &CommitEventData) -> Result<SequenceNumber, DbError> {
async fn insert_commit_event(&self, data: &CommitEventData) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool
.send(MetastoreRequest::Event(EventRequest::InsertCommitEvent {
data: data.clone(),
tx,
}))?;
recv(rx).await
recv(rx).await.map(|_: SequenceNumber| ())
}
async fn insert_identity_event(
&self,
did: &Did,
handle: Option<&Handle>,
) -> Result<SequenceNumber, DbError> {
) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool
.send(MetastoreRequest::Event(EventRequest::InsertIdentityEvent {
@@ -496,14 +496,10 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
handle: handle.cloned(),
tx,
}))?;
recv(rx).await
recv(rx).await.map(|_: SequenceNumber| ())
}
async fn insert_account_event(
&self,
did: &Did,
status: AccountStatus,
) -> Result<SequenceNumber, DbError> {
async fn insert_account_event(&self, did: &Did, status: AccountStatus) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool
.send(MetastoreRequest::Event(EventRequest::InsertAccountEvent {
@@ -511,7 +507,7 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
status,
tx,
}))?;
recv(rx).await
recv(rx).await.map(|_: SequenceNumber| ())
}
async fn insert_sync_event(
@@ -520,7 +516,7 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
commit_cid: &CidLink,
rev: Option<&str>,
commit_bytes: &[u8],
) -> Result<SequenceNumber, DbError> {
) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool
.send(MetastoreRequest::Event(EventRequest::InsertSyncEvent {
@@ -530,7 +526,7 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
commit_bytes: commit_bytes.to_vec(),
tx,
}))?;
recv(rx).await
recv(rx).await.map(|_: SequenceNumber| ())
}
async fn insert_genesis_commit_event(
@@ -541,7 +537,7 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
rev: &str,
commit_bytes: &[u8],
mst_root_bytes: &[u8],
) -> Result<SequenceNumber, DbError> {
) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool.send(MetastoreRequest::Event(
EventRequest::InsertGenesisCommitEvent {
@@ -554,19 +550,14 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
tx,
},
))?;
recv(rx).await
recv(rx).await.map(|_: SequenceNumber| ())
}
async fn delete_sequences_except(
&self,
did: &Did,
keep_seq: SequenceNumber,
) -> Result<(), DbError> {
async fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool.send(MetastoreRequest::Event(
EventRequest::DeleteSequencesExcept {
EventRequest::PurgeDidEventsKeepingLatest {
did: did.clone(),
keep_seq,
tx,
},
))?;
@@ -713,16 +704,6 @@ impl<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
recv(rx).await
}
async fn notify_update(&self, seq: SequenceNumber) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool
.send(MetastoreRequest::Event(EventRequest::NotifyUpdate {
seq,
tx,
}))?;
recv(rx).await
}
async fn import_repo_data(
&self,
user_id: Uuid,
@@ -4898,7 +4879,7 @@ impl<S: StorageIO + 'static> tranquil_db_traits::UserRepository for MetastoreCli
recv(rx).await
}
async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<i64, DbError> {
async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> {
let (tx, rx) = oneshot::channel();
self.pool.send(MetastoreRequest::User(
UserRequest::DeleteAccountWithFirehose {
crates/tranquil-store/src/metastore/commit_ops.rs
+6 -10
View File
@@ -245,7 +245,7 @@ impl<S: StorageIO + 'static> CommitOps<S> {
.serialize()
.map_err(|e| ApplyCommitError::Database(e.to_string()))?;
let (seq, deferred) = self
let (_seq, deferred) = self
.event_ops
.append_commit_event_into_batch(
&mut batch,
@@ -261,7 +261,6 @@ impl<S: StorageIO + 'static> CommitOps<S> {
self.event_ops.complete_broadcast(deferred);
Ok(ApplyCommitResult {
seq: seq.as_i64(),
is_account_active: is_active,
})
}
@@ -477,7 +476,7 @@ mod tests {
use crate::eventlog::{EventLog, EventLogConfig};
use crate::io::RealIO;
use crate::metastore::{Metastore, MetastoreConfig};
use tranquil_db_traits::{CommitEventData, RepoEventType, SequenceNumber};
use tranquil_db_traits::{CommitEventData, RepoEventType};
use tranquil_types::{Handle, Nsid, Rkey};
struct TestHarness {
@@ -599,7 +598,6 @@ mod tests {
};
let result = ops.apply_commit(input).unwrap();
assert!(result.seq > 0);
assert!(result.is_account_active);
let repo = h.metastore.repo_ops().get_repo(user_id).unwrap().unwrap();
@@ -812,8 +810,8 @@ mod tests {
},
};
let result = ops.apply_commit(input).unwrap();
let seq = SequenceNumber::from_raw(result.seq);
ops.apply_commit(input).unwrap();
let seq = ops.event_ops.get_max_seq();
let event = ops.event_ops.get_event_by_seq(seq).unwrap().unwrap();
assert_eq!(event.did, did);
@@ -954,8 +952,7 @@ mod tests {
},
};
let result = ops.apply_commit(input).unwrap();
assert!(result.seq > 0);
ops.apply_commit(input).unwrap();
}
#[test]
@@ -1144,8 +1141,7 @@ mod tests {
},
};
let result = ops.apply_commit(input).unwrap();
assert!(result.seq > 0);
ops.apply_commit(input).unwrap();
metastore.persist().unwrap();
}
crates/tranquil-store/src/metastore/event_ops.rs
+27 -4
View File
@@ -41,10 +41,6 @@ impl<S: StorageIO + 'static> EventOps<S> {
self.bridge.notifier()
}
pub fn notify_update(&self, _seq: SequenceNumber) -> Result<(), DbError> {
Ok(())
}
pub fn insert_commit_event(&self, data: &CommitEventData) -> Result<SequenceNumber, DbError> {
let event = Self::build_commit_event(data);
self.append_and_index(&event, &data.did, data.rev.as_deref())
@@ -382,6 +378,33 @@ impl<S: StorageIO + 'static> EventOps<S> {
Ok(())
}
pub fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError> {
let user_hash = UserHash::from_did(did.as_str());
let prefix = did_events_prefix(user_hash);
let upper = exclusive_upper_bound(prefix.as_slice())
.expect("did_events prefix can never be all-0xFF");
let latest = self
.repo_data
.range(prefix.as_slice()..upper.as_slice())
.map(|guard| {
let (key, _) = guard.into_inner().map_err(fjall_to_db)?;
decode_did_events_seq(key.as_ref())
})
.collect::<Result<Vec<u64>, DbError>>()?
.into_iter()
.max();
match latest {
Some(seq) => {
let keep = i64::try_from(seq)
.map_err(|_| DbError::Query("sequence number out of range".to_owned()))?;
self.delete_sequences_except(did, SequenceNumber::from_raw(keep))
}
None => Ok(()),
}
}
pub fn read_last_applied_cursor(&self) -> Result<Option<u64>, DbError> {
let key = metastore_cursor_key();
match self.repo_data.get(key.as_slice()).map_err(fjall_to_db)? {
crates/tranquil-store/src/metastore/handler.rs
+7 -16
View File
@@ -362,9 +362,8 @@ pub enum EventRequest {
mst_root_bytes: Vec<u8>,
tx: Tx<SequenceNumber>,
},
DeleteSequencesExcept {
PurgeDidEventsKeepingLatest {
did: Did,
keep_seq: SequenceNumber,
tx: Tx<()>,
},
GetMaxSeq {
@@ -393,10 +392,6 @@ pub enum EventRequest {
limit: i64,
tx: Tx<Vec<SequencedEvent>>,
},
NotifyUpdate {
seq: SequenceNumber,
tx: Tx<()>,
},
}
impl EventRequest {
@@ -409,7 +404,7 @@ impl EventRequest {
| Self::InsertAccountEvent { did, .. }
| Self::InsertSyncEvent { did, .. }
| Self::InsertGenesisCommitEvent { did, .. }
| Self::DeleteSequencesExcept { did, .. } => {
| Self::PurgeDidEventsKeepingLatest { did, .. } => {
Routing::Sharded(UserHash::from_did(did.as_str()).raw())
}
Self::GetMaxSeq { .. }
@@ -417,8 +412,7 @@ impl EventRequest {
| Self::GetEventsSinceSeq { .. }
| Self::GetEventsInSeqRange { .. }
| Self::GetEventBySeq { .. }
| Self::GetEventsSinceCursor { .. }
| Self::NotifyUpdate { .. } => Routing::Global,
| Self::GetEventsSinceCursor { .. } => Routing::Global,
}
}
}
@@ -1513,7 +1507,7 @@ pub enum UserRequest {
DeleteAccountWithFirehose {
user_id: Uuid,
did: Did,
tx: Tx<i64>,
tx: Tx<()>,
},
CreatePasswordAccount {
input: CreatePasswordAccountInput,
@@ -2967,8 +2961,8 @@ fn dispatch_event<S: StorageIO + 'static>(state: &HandlerState<S>, req: EventReq
);
let _ = tx.send(result);
}
EventRequest::DeleteSequencesExcept { did, keep_seq, tx } => {
let result = state.event_ops.delete_sequences_except(&did, keep_seq);
EventRequest::PurgeDidEventsKeepingLatest { did, tx } => {
let result = state.event_ops.purge_did_events_keeping_latest(&did);
let _ = tx.send(result);
}
EventRequest::GetMaxSeq { tx } => {
@@ -2997,9 +2991,6 @@ fn dispatch_event<S: StorageIO + 'static>(state: &HandlerState<S>, req: EventReq
EventRequest::GetEventsSinceCursor { cursor, limit, tx } => {
let _ = tx.send(state.event_ops.get_events_since_cursor(cursor, limit));
}
EventRequest::NotifyUpdate { seq, tx } => {
let _ = tx.send(state.event_ops.notify_update(seq));
}
}
}
@@ -5804,7 +5795,7 @@ fn dispatch_user<S: StorageIO + 'static>(state: &HandlerState<S>, req: UserReque
.event_ops
.insert_account_event(&did, AccountStatus::Deleted)
});
let _ = tx.send(result.map(|seq| seq.as_i64()));
let _ = tx.send(result.map(|_| ()));
}
UserRequest::CreatePasswordAccount { input, tx } => {
let result = user.create_password_account(&input).and_then(|result| {
crates/tranquil-store/tests/sim_metastore.rs
+2 -11
View File
@@ -405,13 +405,9 @@ fn sim_crash_recovery_cursor_tracks_last_durable_commit() {
),
};
let result = h.apply_commit(input).unwrap_or_else(|e| {
h.apply_commit(input).unwrap_or_else(|e| {
panic!("seed={seed} commit {commit_idx}: {e:?}")
});
assert!(
result.seq > 0,
"seed={seed} commit {commit_idx} seq must be positive"
);
let new_persisted = match commit_idx <= crash_after {
true => {
@@ -689,12 +685,7 @@ fn sim_handler_pool_shutdown_with_inflight_commits() {
receivers.into_iter().for_each(|(idx, rx)| {
let result = rt.block_on(rx);
match result {
Ok(Ok(commit_result)) => {
assert!(
commit_result.seq > 0,
"seed={seed} idx={idx} commit seq must be positive"
);
}
Ok(Ok(_commit_result)) => {}
Ok(Err(e)) => {
panic!("seed={seed} idx={idx} commit failed: {e:?}");
}
crates/tranquil-sync/src/listener.rs
+47 -78
View File
@@ -1,11 +1,15 @@
use std::sync::atomic::{AtomicI64, Ordering};
use tracing::{debug, error, info, warn};
use std::time::Duration;
use tracing::{error, info, warn};
use tranquil_db_traits::SequenceNumber;
use tranquil_pds::state::AppState;
use tranquil_pds::sync::firehose::SequencedEvent;
static LAST_BROADCAST_SEQ: AtomicI64 = AtomicI64::new(0);
const DRAIN_BATCH_SIZE: i64 = 1000;
const POLL_INTERVAL: Duration = Duration::from_secs(1);
pub async fn start_sequencer_listener(state: AppState) {
let initial_seq = state
.repos
@@ -23,7 +27,7 @@ pub async fn start_sequencer_listener(state: AppState) {
loop {
if let Err(e) = listen_loop(state.clone()).await {
error!("Sequencer listener failed: {}. Restarting in 5s...", e);
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
tokio::time::sleep(Duration::from_secs(5)).await;
}
}
});
@@ -37,88 +41,53 @@ async fn listen_loop(state: AppState) -> anyhow::Result<()> {
.await
.map_err(|e| anyhow::anyhow!("Failed to subscribe to events: {:?}", e))?;
info!("Connected to database and listening for repo updates");
let catchup_start = SequenceNumber::from_raw(LAST_BROADCAST_SEQ.load(Ordering::SeqCst));
let events = state
.repos
.repo
.get_events_since_seq(catchup_start, None)
.await
.map_err(|e| anyhow::anyhow!("Failed to fetch catchup events: {:?}", e))?;
if !events.is_empty() {
info!(
count = events.len(),
from_seq = catchup_start.as_i64(),
"Broadcasting catch-up events"
);
events.into_iter().for_each(|event| {
let seq = event.seq;
let firehose_event = to_firehose_event(event);
let _ = state.firehose_tx.send(firehose_event);
LAST_BROADCAST_SEQ.store(seq.as_i64(), Ordering::SeqCst);
});
let mut last_seq = LAST_BROADCAST_SEQ.load(Ordering::SeqCst);
sequence_and_broadcast(&state, &mut last_seq).await;
loop {
tokio::select! {
received = receiver.recv() => {
if received.is_none() {
return Err(anyhow::anyhow!("Event receiver disconnected"));
}
}
_ = tokio::time::sleep(POLL_INTERVAL) => {}
}
sequence_and_broadcast(&state, &mut last_seq).await;
}
}
async fn sequence_and_broadcast(state: &AppState, last_seq: &mut i64) {
if let Err(e) = state.repos.repo.assign_pending_sequences().await {
warn!("Failed to assign pending firehose sequences: {:?}", e);
}
loop {
let Some(seq_id) = receiver.recv().await else {
return Err(anyhow::anyhow!("Event receiver disconnected"));
};
debug!(seq = seq_id, "Received event notification");
let last_seq = LAST_BROADCAST_SEQ.load(Ordering::SeqCst);
if seq_id <= last_seq {
debug!(
seq = seq_id,
last = last_seq,
"Skipping already-broadcast event"
);
continue;
}
if seq_id > last_seq + 1 {
let gap_events = state
.repos
.repo
.get_events_in_seq_range(
SequenceNumber::from_raw(last_seq),
SequenceNumber::from_raw(seq_id),
)
.await
.unwrap_or_default();
if !gap_events.is_empty() {
debug!(count = gap_events.len(), "Filling sequence gap");
gap_events.into_iter().for_each(|event| {
let seq = event.seq;
let firehose_event = to_firehose_event(event);
let _ = state.firehose_tx.send(firehose_event);
LAST_BROADCAST_SEQ.store(seq.as_i64(), Ordering::SeqCst);
});
}
}
let event = state
let events = match state
.repos
.repo
.get_event_by_seq(SequenceNumber::from_raw(seq_id))
.get_events_since_seq(SequenceNumber::from_raw(*last_seq), Some(DRAIN_BATCH_SIZE))
.await
.ok()
.flatten();
if let Some(event) = event {
let seq = event.seq;
let firehose_event = to_firehose_event(event);
match state.firehose_tx.send(firehose_event) {
Ok(receiver_count) => {
debug!(
seq = seq_id,
receivers = receiver_count,
"Broadcast event to firehose"
);
}
Err(e) => {
warn!(seq = seq_id, error = %e, "Failed to broadcast event (no receivers?)");
}
{
Ok(events) => events,
Err(e) => {
warn!("Sequencer broadcast query failed: {:?}", e);
return;
}
LAST_BROADCAST_SEQ.store(seq.as_i64(), Ordering::SeqCst);
} else {
warn!(
seq = seq_id,
"Received notification but could not find row in repo_seq"
);
};
if events.is_empty() {
return;
}
let batch_len = events.len();
for event in events {
let seq = event.seq.as_i64();
let firehose_event = to_firehose_event(event);
let _ = state.firehose_tx.send(firehose_event);
*last_seq = seq;
LAST_BROADCAST_SEQ.store(seq, Ordering::SeqCst);
}
if (batch_len as i64) < DRAIN_BATCH_SIZE {
return;
}
}
}
migrations/20260529_firehose_outbox_sequencing.sql
+15
View File
@@ -0,0 +1,15 @@
ALTER TABLE repo_seq ADD COLUMN id BIGSERIAL;
ALTER TABLE repo_seq DROP CONSTRAINT repo_seq_pkey;
ALTER TABLE repo_seq ADD PRIMARY KEY (id);
ALTER TABLE repo_seq ALTER COLUMN seq DROP DEFAULT;
ALTER TABLE repo_seq ALTER COLUMN seq DROP NOT NULL;
DROP INDEX IF EXISTS idx_repo_seq_seq;
ALTER TABLE repo_seq ADD CONSTRAINT repo_seq_seq_key UNIQUE (seq);
CREATE SEQUENCE IF NOT EXISTS firehose_seq;
SELECT setval('firehose_seq', (SELECT COALESCE(MAX(seq), 0) + 1 FROM repo_seq), false);
CREATE INDEX idx_repo_seq_unsequenced ON repo_seq (id) WHERE seq IS NULL;