diff --git a/docs/src/content/docs/components/gitmirror.md b/docs/src/content/docs/components/gitmirror.md index f492a049c..8b791bfba 100644 --- a/docs/src/content/docs/components/gitmirror.md +++ b/docs/src/content/docs/components/gitmirror.md @@ -16,8 +16,12 @@ infrastructure behind [bobbin](/components/bobbin/). 1. A knot writes an `org.tangled.git.refs/self` record after every push. 2. Hydrant collects it; gitmirror reads it from hydrant's `/stream` and stores the ref set in Postgres. -3. A sync worker resolves the repo DID's knot from its DID document and - fetches the new objects into a bare repo under `GITMIRROR_SCAN_PATH`. +3. Any other `org.tangled.*` record event whose atproto commit `rev` is + newer than the stored one also queues a resync, so record-only writes + (issues, comments) reach the mirror without a push. +4. A sync worker resolves the repo DID's knot from its DID document and + fetches the new objects into a bare repo under `GITMIRROR_SCAN_PATH`: + the branches and tags from the refs record, plus `refs/cobs/*`. Sync runs with a global limit (`GITMIRROR_SYNC_CONCURRENCY`, default 8), a per-knot limit (`GITMIRROR_SYNC_PER_KNOT_CONCURRENCY`, default 4) and @@ -25,8 +29,8 @@ backoff, so one slow knot doesn't stall the rest. The knot stays authoritative: if the mirror is behind or down, the repo itself is unaffected, but browsing shows stale data or errors. -The hydrant gitmirror reads from must include `org.tangled.git.refs` in -both its collections and its signals. +The hydrant gitmirror reads from must include `org.tangled.*` in its +collections and `org.tangled.git.refs` in its signals. ## API diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-4639c2d4a3d29f03f9145d7352eb9d9751d1ffec1101bf42196c52fff6047a5a.json b/gitmirror/crates/gitmirror-db/.sqlx/query-4639c2d4a3d29f03f9145d7352eb9d9751d1ffec1101bf42196c52fff6047a5a.json new file mode 100644 index 000000000..73e39111b --- /dev/null +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-4639c2d4a3d29f03f9145d7352eb9d9751d1ffec1101bf42196c52fff6047a5a.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "update repos set retry_after_us = $3, updated_at = now()\n where did = $1 and refs_cid = $2 and atproto_rev is not distinct from $4\n and deleted_at is null", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8", + "Text" + ] + }, + "nullable": [] + }, + "hash": "4639c2d4a3d29f03f9145d7352eb9d9751d1ffec1101bf42196c52fff6047a5a" +} diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-d8a554b5420520d584b5b759a04c8b31d1a9c159842af5aa0a76f7a89713057c.json b/gitmirror/crates/gitmirror-db/.sqlx/query-48bc0c0d369583d5cce9f45760338feedf8c3f714eabe06dc1e2aba3dddb0c62.json similarity index 64% rename from gitmirror/crates/gitmirror-db/.sqlx/query-d8a554b5420520d584b5b759a04c8b31d1a9c159842af5aa0a76f7a89713057c.json rename to gitmirror/crates/gitmirror-db/.sqlx/query-48bc0c0d369583d5cce9f45760338feedf8c3f714eabe06dc1e2aba3dddb0c62.json index 00f9ad70f..8063524f5 100644 --- a/gitmirror/crates/gitmirror-db/.sqlx/query-d8a554b5420520d584b5b759a04c8b31d1a9c159842af5aa0a76f7a89713057c.json +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-48bc0c0d369583d5cce9f45760338feedf8c3f714eabe06dc1e2aba3dddb0c62.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "update repos\n set state = 'error', error_msg = $3, retry_count = retry_count + 1,\n retry_after_us = $4, updated_at = now()\n where did = $1 and refs_cid = $2 and deleted_at is null", + "query": "update repos\n set state = 'error', error_msg = $3, retry_count = retry_count + 1,\n retry_after_us = $4, updated_at = now()\n where did = $1 and refs_cid = $2 and atproto_rev is not distinct from $5\n and deleted_at is null", "describe": { "columns": [], "parameters": { @@ -8,10 +8,11 @@ "Text", "Text", "Text", - "Int8" + "Int8", + "Text" ] }, "nullable": [] }, - "hash": "d8a554b5420520d584b5b759a04c8b31d1a9c159842af5aa0a76f7a89713057c" + "hash": "48bc0c0d369583d5cce9f45760338feedf8c3f714eabe06dc1e2aba3dddb0c62" } diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758.json b/gitmirror/crates/gitmirror-db/.sqlx/query-b2e6317ee0c6d9ef282e9d02c5b6b42d7b656676e2465608ff0618b702735270.json similarity index 64% rename from gitmirror/crates/gitmirror-db/.sqlx/query-e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758.json rename to gitmirror/crates/gitmirror-db/.sqlx/query-b2e6317ee0c6d9ef282e9d02c5b6b42d7b656676e2465608ff0618b702735270.json index a71653dbe..a976214ba 100644 --- a/gitmirror/crates/gitmirror-db/.sqlx/query-e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758.json +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-b2e6317ee0c6d9ef282e9d02c5b6b42d7b656676e2465608ff0618b702735270.json @@ -1,10 +1,11 @@ { "db_name": "PostgreSQL", - "query": "update repos\n set state = 'active', error_msg = null, retry_count = 0, retry_after_us = null,\n default_branch_commit = $3, updated_at = now()\n where did = $1 and refs_cid = $2 and deleted_at is null", + "query": "update repos\n set state = 'active', error_msg = null, retry_count = 0, retry_after_us = null,\n default_branch_commit = $3, updated_at = now()\n where did = $1 and refs_cid = $2 and atproto_rev is not distinct from $4\n and deleted_at is null", "describe": { "columns": [], "parameters": { "Left": [ + "Text", "Text", "Text", "Text" @@ -12,5 +13,5 @@ }, "nullable": [] }, - "hash": "e1cb3af8f931c7f43ff26dae9801a4711e0805a771b398f78d8cc43fe15f2758" + "hash": "b2e6317ee0c6d9ef282e9d02c5b6b42d7b656676e2465608ff0618b702735270" } diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-bdcad83ba694785812ced85914595227d765603eb5299393e1ba39af09222fa7.json b/gitmirror/crates/gitmirror-db/.sqlx/query-bdcad83ba694785812ced85914595227d765603eb5299393e1ba39af09222fa7.json new file mode 100644 index 000000000..e3be41c08 --- /dev/null +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-bdcad83ba694785812ced85914595227d765603eb5299393e1ba39af09222fa7.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "update repos\n set atproto_rev = $2, state = 'pending', error_msg = null,\n retry_count = 0, retry_after_us = null, updated_at = now()\n where did = $1 and deleted_at is null\n and (atproto_rev is null or atproto_rev < $2)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "bdcad83ba694785812ced85914595227d765603eb5299393e1ba39af09222fa7" +} diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-b36c6631c70015f445c255ca16d81fad68f90fa21d062351661d2ab5d9c91e8e.json b/gitmirror/crates/gitmirror-db/.sqlx/query-c467065187d676ef4fb8e9d3bdbb6b0a7a33ecf56143fa25e76b2e27df7de3c2.json similarity index 67% rename from gitmirror/crates/gitmirror-db/.sqlx/query-b36c6631c70015f445c255ca16d81fad68f90fa21d062351661d2ab5d9c91e8e.json rename to gitmirror/crates/gitmirror-db/.sqlx/query-c467065187d676ef4fb8e9d3bdbb6b0a7a33ecf56143fa25e76b2e27df7de3c2.json index af85be842..9d09d4262 100644 --- a/gitmirror/crates/gitmirror-db/.sqlx/query-b36c6631c70015f445c255ca16d81fad68f90fa21d062351661d2ab5d9c91e8e.json +++ b/gitmirror/crates/gitmirror-db/.sqlx/query-c467065187d676ef4fb8e9d3bdbb6b0a7a33ecf56143fa25e76b2e27df7de3c2.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select did, refs_cid, default_branch, refs, retry_count\n from repos\n where deleted_at is null\n and state <> 'active'\n and (retry_after_us is null or retry_after_us <= $1)\n order by retry_after_us nulls first, updated_at\n limit $2", + "query": "select did, refs_cid, default_branch, refs, atproto_rev, retry_count\n from repos\n where deleted_at is null\n and state <> 'active'\n and (retry_after_us is null or retry_after_us <= $1)\n order by retry_after_us nulls first, updated_at\n limit $2", "describe": { "columns": [ { @@ -49,6 +49,17 @@ }, { "ordinal": 4, + "name": "atproto_rev", + "type_info": "Text", + "origin": { + "Table": { + "table": "repos", + "name": "atproto_rev" + } + } + }, + { + "ordinal": 5, "name": "retry_count", "type_info": "Int4", "origin": { @@ -70,8 +81,9 @@ false, false, false, + true, false ] }, - "hash": "b36c6631c70015f445c255ca16d81fad68f90fa21d062351661d2ab5d9c91e8e" + "hash": "c467065187d676ef4fb8e9d3bdbb6b0a7a33ecf56143fa25e76b2e27df7de3c2" } diff --git a/gitmirror/crates/gitmirror-db/.sqlx/query-cb4b72b818a1f2922826955d4ec7948bd99a40d342d5687dd8c1c8dd137a2508.json b/gitmirror/crates/gitmirror-db/.sqlx/query-cb4b72b818a1f2922826955d4ec7948bd99a40d342d5687dd8c1c8dd137a2508.json deleted file mode 100644 index 2edb39e77..000000000 --- a/gitmirror/crates/gitmirror-db/.sqlx/query-cb4b72b818a1f2922826955d4ec7948bd99a40d342d5687dd8c1c8dd137a2508.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "update repos set retry_after_us = $3, updated_at = now()\n where did = $1 and refs_cid = $2 and deleted_at is null", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Int8" - ] - }, - "nullable": [] - }, - "hash": "cb4b72b818a1f2922826955d4ec7948bd99a40d342d5687dd8c1c8dd137a2508" -} diff --git a/gitmirror/crates/gitmirror-db/migrations/20261019000000_atproto_rev.sql b/gitmirror/crates/gitmirror-db/migrations/20261019000000_atproto_rev.sql new file mode 100644 index 000000000..ffd6c335a --- /dev/null +++ b/gitmirror/crates/gitmirror-db/migrations/20261019000000_atproto_rev.sql @@ -0,0 +1 @@ +alter table repos add column atproto_rev text; diff --git a/gitmirror/crates/gitmirror-db/src/lib.rs b/gitmirror/crates/gitmirror-db/src/lib.rs index 27ffca80a..f965e7cba 100644 --- a/gitmirror/crates/gitmirror-db/src/lib.rs +++ b/gitmirror/crates/gitmirror-db/src/lib.rs @@ -80,6 +80,7 @@ pub struct RepoRow { pub did: Did, pub refs_cid: String, pub refs: GitRefs, + pub atproto_rev: Option, pub retry_count: i32, } @@ -155,6 +156,23 @@ impl Db { Ok(done.rows_affected() > 0) } + /// Records a newer atproto commit rev for the account, queueing a resync. Returns whether + /// the rev moved forward; repos with no refs record yet are left alone. + pub async fn bump_rev(&self, did: &Did, rev: &str) -> Result { + let done = sqlx::query!( + "update repos + set atproto_rev = $2, state = 'pending', error_msg = null, + retry_count = 0, retry_after_us = null, updated_at = now() + where did = $1 and deleted_at is null + and (atproto_rev is null or atproto_rev < $2)", + did.as_str(), + rev, + ) + .execute(&self.0) + .await?; + Ok(done.rows_affected() > 0) + } + pub async fn mark_deleted(&self, did: &Did) -> Result<(), DbError> { sqlx::query!( "update repos set deleted_at = now(), updated_at = now() @@ -169,7 +187,7 @@ impl Db { /// Repos due for a git resync, longest-waiting first. pub async fn pending_repos(&self, now_us: i64, limit: i64) -> Result, DbError> { let rows = sqlx::query!( - "select did, refs_cid, default_branch, refs, retry_count + "select did, refs_cid, default_branch, refs, atproto_rev, retry_count from repos where deleted_at is null and state <> 'active' @@ -194,6 +212,7 @@ impl Db { default_branch: r.default_branch.into(), refs: Bytes::from(r.refs), }, + atproto_rev: r.atproto_rev, retry_count: r.retry_count, }) }) @@ -229,59 +248,59 @@ impl Db { pub async fn mark_synced( &self, - did: &Did, - refs_cid: &str, + row: &RepoRow, default_branch_commit: Option<&str>, ) -> Result<(), DbError> { + let (did, refs_cid) = (&row.did, &row.refs_cid); let done = sqlx::query!( "update repos set state = 'active', error_msg = null, retry_count = 0, retry_after_us = null, default_branch_commit = $3, updated_at = now() - where did = $1 and refs_cid = $2 and deleted_at is null", + where did = $1 and refs_cid = $2 and atproto_rev is not distinct from $4 + and deleted_at is null", did.as_str(), refs_cid, default_branch_commit, + row.atproto_rev, ) .execute(&self.0) .await?; if done.rows_affected() > 0 { debug!(%did, refs_cid, "synced"); } else { - debug!(%did, refs_cid, "refs changed mid-sync, dropped the stale success"); + debug!(%did, refs_cid, "refs or rev changed mid-sync, dropped the stale success"); } Ok(()) } - pub async fn mark_deferred( - &self, - did: &Did, - refs_cid: &str, - retry_after_us: i64, - ) -> Result<(), DbError> { + pub async fn mark_deferred(&self, row: &RepoRow, retry_after_us: i64) -> Result<(), DbError> { + let (did, refs_cid) = (&row.did, &row.refs_cid); let done = sqlx::query!( "update repos set retry_after_us = $3, updated_at = now() - where did = $1 and refs_cid = $2 and deleted_at is null", + where did = $1 and refs_cid = $2 and atproto_rev is not distinct from $4 + and deleted_at is null", did.as_str(), refs_cid, retry_after_us, + row.atproto_rev, ) .execute(&self.0) .await?; if done.rows_affected() > 0 { debug!(%did, refs_cid, retry_after_us, "knot busy, deferred"); } else { - debug!(%did, refs_cid, "refs changed mid-sync, dropped the stale defer"); + debug!(%did, refs_cid, "refs or rev changed mid-sync, dropped the stale defer"); } Ok(()) } pub async fn mark_sync_failed( &self, - did: &Did, - refs_cid: &str, + row: &RepoRow, error: &str, retry_after_us: i64, ) -> Result<(), DbError> { + let (did, refs_cid) = (&row.did, &row.refs_cid); // postgres `text` rejects NUL let error = if error.contains('\0') { Cow::Owned(error.replace('\0', "")) @@ -292,18 +311,20 @@ impl Db { "update repos set state = 'error', error_msg = $3, retry_count = retry_count + 1, retry_after_us = $4, updated_at = now() - where did = $1 and refs_cid = $2 and deleted_at is null", + where did = $1 and refs_cid = $2 and atproto_rev is not distinct from $5 + and deleted_at is null", did.as_str(), refs_cid, &*error, retry_after_us, + row.atproto_rev, ) .execute(&self.0) .await?; if done.rows_affected() > 0 { warn!(%did, refs_cid, retry_after_us, error = %error, "sync failed, backing off"); } else { - debug!(%did, refs_cid, "refs changed mid-sync, dropped the stale failure"); + debug!(%did, refs_cid, "refs or rev changed mid-sync, dropped the stale failure"); } Ok(()) } diff --git a/gitmirror/crates/gitmirror-ingest/src/lib.rs b/gitmirror/crates/gitmirror-ingest/src/lib.rs index 33dd63eb6..05ddecbe8 100644 --- a/gitmirror/crates/gitmirror-ingest/src/lib.rs +++ b/gitmirror/crates/gitmirror-ingest/src/lib.rs @@ -7,6 +7,7 @@ use bobbin_runtime::{Clock, WsConn, WsMessage, WsTransport}; use gitmirror_db::{Db, DbError, GitRefs, RefsUpsert}; use jacquard_common::types::collection::Collection as _; use jacquard_common::types::did::Did; +use jacquard_common::types::tid::Tid; use lexicons::org_tangled::git::refs::{Refs, RefsRecord}; use serde::Deserialize; use serde_json::value::RawValue; @@ -17,6 +18,7 @@ use url::Url; const REFS_COLLECTION: &str = RefsRecord::NSID; const REFS_RKEY: &str = "self"; +const TANGLED_PREFIX: &str = "org.tangled."; pub const DEFAULT_CHECKPOINT_INTERVAL: Duration = Duration::from_secs(10); @@ -83,6 +85,10 @@ struct RecordFrame { record: Option>, #[serde(default)] cid: Option, + /// The account's atproto commit rev. Kept as a string so a malformed one skips this + /// record instead of failing the whole frame. + #[serde(default)] + rev: Option, } #[derive(Debug, Deserialize)] @@ -233,56 +239,24 @@ async fn handle(frame: Frame, rt: &IngestRuntime) -> Result<(), IngestError> { let cursor = frame.id; if let Some(record) = frame.record { - if record.collection != REFS_COLLECTION || record.rkey != REFS_RKEY { + if !record.collection.starts_with(TANGLED_PREFIX) { return Ok(()); } + if record.collection == REFS_COLLECTION && record.rkey == REFS_RKEY { + handle_refs(&record, cursor, rt).await?; + } let did = &record.did; - match record.action.as_str() { - "create" | "update" => { - let Some(raw) = record.record.as_ref() else { - warn!(%did, cursor, "refs create/update with no body, skipping"); - return Ok(()); - }; - let Some(cid) = record.cid.as_ref() else { - warn!(%did, cursor, "refs create/update with no cid, skipping"); - return Ok(()); - }; - let refs: Refs = match serde_json::from_str(raw.get()) { - Ok(refs) => refs, - Err(e) => { - warn!(%did, cursor, error = %e, "undecodable refs record, skipping"); - return Ok(()); - } - }; - let git_refs = GitRefs { - default_branch: (&*refs.default_branch).into(), - refs: refs.refs, - }; - let changed = rt - .db - .upsert_refs(RefsUpsert { - did, - refs_cid: cid, - refs: &git_refs, - }) - .await?; - if changed { - debug!(%did, cursor, "stored refs"); - } else { - debug!(%did, cursor, "refs unchanged, skipped write"); - } - Ok(()) - } - "delete" => { - rt.db.mark_deleted(did).await?; - info!(%did, cursor, "refs record deleted, tombstoned row"); - Ok(()) - } - other => { - debug!(%did, cursor, action = other, "ignoring unknown record action"); - Ok(()) - } + let Some(rev) = record.rev.as_deref() else { + return Ok(()); + }; + if Tid::new(rev).is_err() { + warn!(%did, cursor, rev, "malformed rev, skipping"); + return Ok(()); + } + if rt.db.bump_rev(did, rev).await? { + debug!(%did, cursor, rev, "atproto rev moved, queued resync"); } + Ok(()) } else if let Some(account) = frame.account && !account.active { @@ -294,3 +268,57 @@ async fn handle(frame: Frame, rt: &IngestRuntime) -> Result<(), IngestError> { Ok(()) } } + +async fn handle_refs( + record: &RecordFrame, + cursor: u64, + rt: &IngestRuntime, +) -> Result<(), IngestError> { + let did = &record.did; + match record.action.as_str() { + "create" | "update" => { + let Some(raw) = record.record.as_ref() else { + warn!(%did, cursor, "refs create/update with no body, skipping"); + return Ok(()); + }; + let Some(cid) = record.cid.as_ref() else { + warn!(%did, cursor, "refs create/update with no cid, skipping"); + return Ok(()); + }; + let refs: Refs = match serde_json::from_str(raw.get()) { + Ok(refs) => refs, + Err(e) => { + warn!(%did, cursor, error = %e, "undecodable refs record, skipping"); + return Ok(()); + } + }; + let git_refs = GitRefs { + default_branch: (&*refs.default_branch).into(), + refs: refs.refs, + }; + let changed = rt + .db + .upsert_refs(RefsUpsert { + did, + refs_cid: cid, + refs: &git_refs, + }) + .await?; + if changed { + debug!(%did, cursor, "stored refs"); + } else { + debug!(%did, cursor, "refs unchanged, skipped write"); + } + Ok(()) + } + "delete" => { + rt.db.mark_deleted(did).await?; + info!(%did, cursor, "refs record deleted, tombstoned row"); + Ok(()) + } + other => { + debug!(%did, cursor, action = other, "ignoring unknown record action"); + Ok(()) + } + } +} diff --git a/gitmirror/crates/gitmirror-sync/src/lib.rs b/gitmirror/crates/gitmirror-sync/src/lib.rs index 849ce1b82..f35943e66 100644 --- a/gitmirror/crates/gitmirror-sync/src/lib.rs +++ b/gitmirror/crates/gitmirror-sync/src/lib.rs @@ -199,26 +199,19 @@ impl BackfillWorker { default_branch_commit, }) => { let commit = default_branch_commit.map(|oid| oid.to_string()); - let res = db - .mark_synced(&row.did, &row.refs_cid, commit.as_deref()) - .await; + let res = db.mark_synced(&row, commit.as_deref()).await; if res.is_ok() { index_notify.notify_one(); } res } Ok(task::TaskDisposition::Deferred) => { - db.mark_deferred( - &row.did, - &row.refs_cid, - now_us(&*clock) + DEFER_HOLD_US, - ) - .await + db.mark_deferred(&row, now_us(&*clock) + DEFER_HOLD_US) + .await } Err(e) => { db.mark_sync_failed( - &row.did, - &row.refs_cid, + &row, &e.to_string(), now_us(&*clock) + backoff_us(row.retry_count), ) diff --git a/nix/modules/knotmirror.nix b/nix/modules/knotmirror.nix index a84a6dad3..863a014b4 100644 --- a/nix/modules/knotmirror.nix +++ b/nix/modules/knotmirror.nix @@ -40,7 +40,7 @@ in hydrantUrl = mkOption { type = types.str; example = "http://127.0.0.1:3000"; - description = "Base URL of the hydrant whose /stream feeds gitmirror. It must include org.tangled.git.refs in its collections and signals"; + description = "Base URL of the hydrant whose /stream feeds gitmirror. It must include org.tangled.* in its collections and org.tangled.git.refs in its signals"; }; logging = {