Spawn the background compaction task.
(
registry: Arc<StreamRegistry>,
router: Arc<CdcRouter>,
mut shutdown: watch::Receiver<bool>,
)
| 20 | |
| 21 | /// Spawn the background compaction task. |
| 22 | pub fn spawn_compaction_task( |
| 23 | registry: Arc<StreamRegistry>, |
| 24 | router: Arc<CdcRouter>, |
| 25 | mut shutdown: watch::Receiver<bool>, |
| 26 | ) -> tokio::task::JoinHandle<()> { |
| 27 | tokio::spawn(async move { |
| 28 | debug!("CDC compaction task started"); |
| 29 | |
| 30 | loop { |
| 31 | tokio::select! { |
| 32 | _ = tokio::time::sleep(COMPACTION_INTERVAL) => {} |
| 33 | _ = shutdown.changed() => { |
| 34 | if *shutdown.borrow() { |
| 35 | debug!("CDC compaction task shutting down"); |
| 36 | return; |
| 37 | } |
| 38 | } |
| 39 | } |
| 40 | |
| 41 | if *shutdown.borrow() { |
| 42 | return; |
| 43 | } |
| 44 | |
| 45 | // Scan all streams with compaction enabled, grouped by tenant. |
| 46 | // Round-robin across tenants: process one stream per tenant in |
| 47 | // rotation so a tenant with many streams doesn't starve others. |
| 48 | let stats = router.buffer_stats(); |
| 49 | |
| 50 | // Group by tenant_id. |
| 51 | let mut by_tenant: std::collections::HashMap<u64, Vec<_>> = |
| 52 | std::collections::HashMap::new(); |
| 53 | for stat in stats { |
| 54 | by_tenant.entry(stat.tenant_id).or_default().push(stat); |
| 55 | } |
| 56 | |
| 57 | // Interleave: process one stream from each tenant, then repeat. |
| 58 | let max_streams = by_tenant.values().map(|v| v.len()).max().unwrap_or(0); |
| 59 | for round in 0..max_streams { |
| 60 | for streams in by_tenant.values() { |
| 61 | let Some(stat) = streams.get(round) else { |
| 62 | continue; |
| 63 | }; |
| 64 | let def = registry.get(stat.tenant_id, &stat.stream_name); |
| 65 | let def = match def { |
| 66 | Some(d) if d.compaction.enabled => d, |
| 67 | _ => continue, |
| 68 | }; |
| 69 | |
| 70 | if let Some(buffer) = router.get_buffer(stat.tenant_id, &stat.stream_name) { |
| 71 | let removed = buffer.compact( |
| 72 | &def.compaction.key_field, |
| 73 | def.compaction.tombstone_grace_secs, |
| 74 | ); |
| 75 | if removed > 0 { |
| 76 | debug!( |
| 77 | stream = %stat.stream_name, |
| 78 | tenant = stat.tenant_id, |
| 79 | removed, |
no test coverage detected