(state: &Arc<SharedState>, entry: &Entry)
| 85 | } |
| 86 | |
| 87 | async fn run_one(state: &Arc<SharedState>, entry: &Entry) { |
| 88 | let audit_ms = entry.retention.audit_retain_ms; |
| 89 | if audit_ms == 0 { |
| 90 | return; // "retain forever" — no purge. |
| 91 | } |
| 92 | let now_ms = std::time::SystemTime::now() |
| 93 | .duration_since(std::time::UNIX_EPOCH) |
| 94 | .unwrap_or_default() |
| 95 | .as_millis() as i64; |
| 96 | let cutoff_system_ms = now_ms.saturating_sub(audit_ms as i64); |
| 97 | |
| 98 | let tenant_id = entry.tenant_id; |
| 99 | let plan = match entry.engine { |
| 100 | BitemporalEngineKind::EdgeStore => PhysicalPlan::Meta(MetaOp::TemporalPurgeEdgeStore { |
| 101 | tenant_id: tenant_id.as_u64(), |
| 102 | collection: entry.collection.clone(), |
| 103 | cutoff_system_ms, |
| 104 | }), |
| 105 | BitemporalEngineKind::DocumentStrict => { |
| 106 | PhysicalPlan::Meta(MetaOp::TemporalPurgeDocumentStrict { |
| 107 | tenant_id: tenant_id.as_u64(), |
| 108 | collection: entry.collection.clone(), |
| 109 | cutoff_system_ms, |
| 110 | }) |
| 111 | } |
| 112 | BitemporalEngineKind::Columnar => PhysicalPlan::Meta(MetaOp::TemporalPurgeColumnar { |
| 113 | tenant_id: tenant_id.as_u64(), |
| 114 | collection: entry.collection.clone(), |
| 115 | cutoff_system_ms, |
| 116 | }), |
| 117 | BitemporalEngineKind::Crdt => PhysicalPlan::Meta(MetaOp::TemporalPurgeCrdt { |
| 118 | tenant_id: tenant_id.as_u64(), |
| 119 | collection: entry.collection.clone(), |
| 120 | cutoff_system_ms, |
| 121 | }), |
| 122 | // For Array entries `entry.collection` carries the array_id (see |
| 123 | // hydration in main.rs — registered under the catalog `name` field). |
| 124 | BitemporalEngineKind::Array => PhysicalPlan::Meta(MetaOp::TemporalPurgeArray { |
| 125 | tenant_id: tenant_id.as_u64(), |
| 126 | array_id: entry.collection.clone(), |
| 127 | cutoff_system_ms, |
| 128 | }), |
| 129 | }; |
| 130 | |
| 131 | match crate::control::server::pgwire::ddl::sync_dispatch::dispatch_async( |
| 132 | state, |
| 133 | tenant_id, |
| 134 | &entry.collection, |
| 135 | plan, |
| 136 | Duration::from_secs(DISPATCH_DEADLINE_SECS), |
| 137 | ) |
| 138 | .await |
| 139 | { |
| 140 | Ok(payload) => { |
| 141 | let purged = parse_count_from_payload(entry.engine, &payload); |
| 142 | if purged > 0 |
| 143 | && let Err(e) = state.wal.append_temporal_purge( |
| 144 | tenant_id, |
no test coverage detected