|
| 1 | +use std::collections::HashSet; |
| 2 | +use std::path::PathBuf; |
| 3 | +use std::sync::Arc; |
| 4 | +use tokio::sync::watch; |
| 5 | + |
| 6 | +use crate::config::Config; |
| 7 | +use crate::db::Db; |
| 8 | + |
| 9 | +/// How often to run a sweep pass. |
| 10 | +const SWEEP_INTERVAL_SECS: u64 = 3600; |
| 11 | + |
| 12 | +/// Maximum repos to process per pass — prevents the sweep from becoming |
| 13 | +/// the O(repos) amplification the admission-control work exists to prevent. |
| 14 | +const REPOS_PER_PASS: usize = 100; |
| 15 | + |
| 16 | +/// Spawn the periodic reconciliation sweep background task. |
| 17 | +pub fn spawn( |
| 18 | + db: Arc<Db>, |
| 19 | + config: Arc<Config>, |
| 20 | + http_client: Arc<reqwest::Client>, |
| 21 | + node_keypair: Arc<gitlawb_core::identity::Keypair>, |
| 22 | + node_did: gitlawb_core::did::Did, |
| 23 | + mut shutdown_rx: watch::Receiver<bool>, |
| 24 | +) { |
| 25 | + tokio::spawn(async move { |
| 26 | + let node_seed = *node_keypair.to_seed(); |
| 27 | + let mut cursor = 0usize; |
| 28 | + |
| 29 | + loop { |
| 30 | + tokio::select! { |
| 31 | + _ = tokio::time::sleep(std::time::Duration::from_secs(SWEEP_INTERVAL_SECS)) => { |
| 32 | + let start = std::time::Instant::now(); |
| 33 | + match run_pass(&db, &config, &http_client, &node_seed, &node_did, &mut cursor).await { |
| 34 | + Ok((count, gaps, filled)) => { |
| 35 | + tracing::info!( |
| 36 | + repos = count, |
| 37 | + gaps_found = gaps, |
| 38 | + gaps_filled = filled, |
| 39 | + elapsed_ms = start.elapsed().as_millis() as u64, |
| 40 | + "reconciliation sweep pass complete" |
| 41 | + ); |
| 42 | + } |
| 43 | + Err(e) => { |
| 44 | + tracing::warn!(err = %e, "reconciliation sweep pass failed"); |
| 45 | + } |
| 46 | + } |
| 47 | + } |
| 48 | + _ = shutdown_rx.changed() => { |
| 49 | + if *shutdown_rx.borrow() { |
| 50 | + tracing::info!("reconciliation sweep: shutdown signal received, exiting"); |
| 51 | + return; |
| 52 | + } |
| 53 | + } |
| 54 | + } |
| 55 | + } |
| 56 | + }); |
| 57 | +} |
| 58 | + |
| 59 | +/// Run one sweep pass. Returns `(repos_scanned, gaps_found, gaps_filled)`. |
| 60 | +async fn run_pass( |
| 61 | + db: &Db, |
| 62 | + config: &Config, |
| 63 | + http_client: &reqwest::Client, |
| 64 | + node_seed: &[u8; 32], |
| 65 | + node_did: &gitlawb_core::did::Did, |
| 66 | + cursor: &mut usize, |
| 67 | +) -> anyhow::Result<(usize, usize, usize)> { |
| 68 | + let all = db.list_all_repos().await?; |
| 69 | + if all.is_empty() { |
| 70 | + *cursor = 0; |
| 71 | + return Ok((0, 0, 0)); |
| 72 | + } |
| 73 | + |
| 74 | + let start = *cursor; |
| 75 | + let end = (start + REPOS_PER_PASS).min(all.len()); |
| 76 | + let batch = &all[start..end]; |
| 77 | + *cursor = if end >= all.len() { 0 } else { end }; |
| 78 | + |
| 79 | + let mut total_gaps_found = 0usize; |
| 80 | + let mut total_gaps_filled = 0usize; |
| 81 | + |
| 82 | + for repo in batch { |
| 83 | + let repo_slug = format!( |
| 84 | + "{}/{}", |
| 85 | + crate::db::normalize_owner_key(&repo.owner_did), |
| 86 | + repo.name |
| 87 | + ); |
| 88 | + |
| 89 | + let disk = PathBuf::from(&repo.disk_path); |
| 90 | + if !disk.exists() { |
| 91 | + tracing::warn!(repo = %repo_slug, "disk path missing, skipping"); |
| 92 | + continue; |
| 93 | + } |
| 94 | + |
| 95 | + let rules = match db.list_visibility_rules(&repo.id).await { |
| 96 | + Ok(r) => r, |
| 97 | + Err(e) => { |
| 98 | + tracing::warn!(repo = %repo_slug, err = %e, "visibility rules fetch failed, skipping"); |
| 99 | + continue; |
| 100 | + } |
| 101 | + }; |
| 102 | + |
| 103 | + if !crate::visibility::listable_at_root(&rules, repo.is_public, &repo.owner_did, None) { |
| 104 | + continue; |
| 105 | + } |
| 106 | + |
| 107 | + let has_path_scoped = crate::git::visibility_pack::has_path_scoped_rule(&rules); |
| 108 | + |
| 109 | + let disk_clone = disk.clone(); |
| 110 | + let owner_clone = repo.owner_did.clone(); |
| 111 | + let rules_clone = rules.clone(); |
| 112 | + let is_public = repo.is_public; |
| 113 | + let replicable = tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<String>> { |
| 114 | + let all_objs = crate::git::push_delta::list_all_objects(&disk_clone)?; |
| 115 | + if all_objs.is_empty() { |
| 116 | + return Ok(Vec::new()); |
| 117 | + } |
| 118 | + let allowed = if has_path_scoped { |
| 119 | + crate::git::visibility_pack::replicable_blob_set( |
| 120 | + &disk_clone, |
| 121 | + &rules_clone, |
| 122 | + is_public, |
| 123 | + &owner_clone, |
| 124 | + )? |
| 125 | + } else { |
| 126 | + HashSet::new() |
| 127 | + }; |
| 128 | + let all_blobs = crate::git::push_delta::all_blob_oids(&disk_clone)?; |
| 129 | + Ok(crate::git::visibility_pack::replicable_objects_fail_closed( |
| 130 | + all_objs, &allowed, &all_blobs, |
| 131 | + )) |
| 132 | + }) |
| 133 | + .await; |
| 134 | + |
| 135 | + let object_list = match replicable { |
| 136 | + Ok(Ok(list)) => list, |
| 137 | + Ok(Err(e)) => { |
| 138 | + tracing::warn!(repo = %repo_slug, err = %e, "full-scan failed, skipping"); |
| 139 | + continue; |
| 140 | + } |
| 141 | + Err(e) => { |
| 142 | + tracing::warn!(repo = %repo_slug, err = %e, "full-scan task panicked, skipping"); |
| 143 | + continue; |
| 144 | + } |
| 145 | + }; |
| 146 | + |
| 147 | + if object_list.is_empty() { |
| 148 | + continue; |
| 149 | + } |
| 150 | + |
| 151 | + let mut repo_gaps = 0usize; |
| 152 | + for sha in &object_list { |
| 153 | + match db.is_pinned(sha).await { |
| 154 | + Ok(true) => {} |
| 155 | + _ => repo_gaps += 1, |
| 156 | + } |
| 157 | + } |
| 158 | + |
| 159 | + if repo_gaps == 0 { |
| 160 | + continue; |
| 161 | + } |
| 162 | + |
| 163 | + total_gaps_found += repo_gaps; |
| 164 | + |
| 165 | + let pinned_ipfs = |
| 166 | + crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, object_list.clone(), db) |
| 167 | + .await; |
| 168 | + |
| 169 | + let _pinned_pinata = crate::pinata::pin_new_objects( |
| 170 | + http_client, |
| 171 | + &config.pinata_upload_url, |
| 172 | + &config.pinata_jwt, |
| 173 | + &disk, |
| 174 | + object_list, |
| 175 | + db, |
| 176 | + ) |
| 177 | + .await; |
| 178 | + |
| 179 | + let repo_filled = pinned_ipfs.len(); |
| 180 | + total_gaps_filled += repo_filled; |
| 181 | + |
| 182 | + crate::metrics::record_reconciliation_gaps_found(repo_gaps as u64); |
| 183 | + crate::metrics::record_reconciliation_gaps_filled(repo_filled as u64); |
| 184 | + |
| 185 | + if repo_filled > 0 { |
| 186 | + tracing::info!( |
| 187 | + repo = %repo_slug, |
| 188 | + gaps = repo_gaps, |
| 189 | + filled = repo_filled, |
| 190 | + "reconciliation sweep filled gaps" |
| 191 | + ); |
| 192 | + } |
| 193 | + |
| 194 | + if has_path_scoped && !config.ipfs_api.is_empty() { |
| 195 | + let p = disk.clone(); |
| 196 | + let owner = repo.owner_did.clone(); |
| 197 | + let r = rules.clone(); |
| 198 | + let is_public_2 = repo.is_public; |
| 199 | + let recipients = tokio::task::spawn_blocking(move || { |
| 200 | + crate::git::visibility_pack::withheld_blob_recipients(&p, &r, is_public_2, &owner) |
| 201 | + }) |
| 202 | + .await; |
| 203 | + |
| 204 | + match recipients { |
| 205 | + Ok(Ok(rec)) if !rec.is_empty() => { |
| 206 | + let sealed = crate::encrypted_pin::encrypt_and_pin( |
| 207 | + &config.ipfs_api, |
| 208 | + &disk, |
| 209 | + db, |
| 210 | + &repo.id, |
| 211 | + node_seed, |
| 212 | + &rec, |
| 213 | + ) |
| 214 | + .await; |
| 215 | + if !sealed.is_empty() && !config.irys_url.is_empty() { |
| 216 | + let owner_short = crate::db::normalize_owner_key(&repo.owner_did); |
| 217 | + let slug = format!("{}/{}", owner_short, repo.name); |
| 218 | + let ts = chrono::Utc::now().to_rfc3339(); |
| 219 | + let node_did_str = node_did.to_string(); |
| 220 | + let blobs: Vec<(String, String)> = sealed; |
| 221 | + let manifest = crate::arweave::EncryptedManifest { |
| 222 | + repo: &slug, |
| 223 | + owner_did: &repo.owner_did, |
| 224 | + node_did: &node_did_str, |
| 225 | + timestamp: &ts, |
| 226 | + blobs: &blobs, |
| 227 | + }; |
| 228 | + if let Err(e) = crate::arweave::anchor_encrypted_manifest( |
| 229 | + http_client, |
| 230 | + &config.irys_url, |
| 231 | + &manifest, |
| 232 | + ) |
| 233 | + .await |
| 234 | + { |
| 235 | + tracing::warn!( |
| 236 | + repo = %slug, |
| 237 | + err = %e, |
| 238 | + "encrypted manifest anchor failed (best-effort)" |
| 239 | + ); |
| 240 | + } |
| 241 | + } |
| 242 | + } |
| 243 | + Ok(Ok(_)) => {} |
| 244 | + Ok(Err(e)) => { |
| 245 | + tracing::warn!( |
| 246 | + repo = %repo_slug, |
| 247 | + err = %e, |
| 248 | + "withheld_blob_recipients failed, skipping encrypted pin" |
| 249 | + ); |
| 250 | + } |
| 251 | + Err(e) => { |
| 252 | + tracing::warn!( |
| 253 | + repo = %repo_slug, |
| 254 | + err = %e, |
| 255 | + "withheld_blob_recipients task panicked, skipping encrypted pin" |
| 256 | + ); |
| 257 | + } |
| 258 | + } |
| 259 | + } |
| 260 | + } |
| 261 | + |
| 262 | + Ok((batch.len(), total_gaps_found, total_gaps_filled)) |
| 263 | +} |
0 commit comments