Archive ephemeral oplogs from a sweep instead of a scheduled action

golem-debugging-service/config/debug-worker-executor.sample.env

@@ -89,6 +89,15 @@ GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MIN_DELAY="100ms"
 GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MULTIPLIER=3.0
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__TYPE="EveryNInvocation"
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__CONFIG__COUNT=10
+GOLEM__OPLOG__SWEEP__ENABLED=true
+GOLEM__OPLOG__SWEEP__INTERVAL="1m"
+GOLEM__OPLOG__SWEEP__MAX_ARCHIVES_PER_TICK=256
+GOLEM__OPLOG__SWEEP__MAX_BACKOFF_INTERVALS=8
+GOLEM__OPLOG__SWEEP__MAX_CONCURRENCY=4
+GOLEM__OPLOG__SWEEP__MAX_SCANNED_PER_TICK=4096
+GOLEM__OPLOG__SWEEP__MAX_TICK_DURATION="30s"
+GOLEM__OPLOG__SWEEP__MAX_TRACKED_AGENTS=100000
+GOLEM__OPLOG__SWEEP__PAGE_SIZE=128
 GOLEM__PUBLIC_WORKER_API__CONNECT_TIMEOUT="10s"
 GOLEM__PUBLIC_WORKER_API__HOST="localhost"
 GOLEM__PUBLIC_WORKER_API__HTTP2_KEEP_ALIVE_INTERVAL="10s"
@@ -282,6 +291,15 @@ GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MIN_DELAY="100ms"
 GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MULTIPLIER=3.0
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__TYPE="EveryNInvocation"
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__CONFIG__COUNT=10
+GOLEM__OPLOG__SWEEP__ENABLED=true
+GOLEM__OPLOG__SWEEP__INTERVAL="1m"
+GOLEM__OPLOG__SWEEP__MAX_ARCHIVES_PER_TICK=256
+GOLEM__OPLOG__SWEEP__MAX_BACKOFF_INTERVALS=8
+GOLEM__OPLOG__SWEEP__MAX_CONCURRENCY=4
+GOLEM__OPLOG__SWEEP__MAX_SCANNED_PER_TICK=4096
+GOLEM__OPLOG__SWEEP__MAX_TICK_DURATION="30s"
+GOLEM__OPLOG__SWEEP__MAX_TRACKED_AGENTS=100000
+GOLEM__OPLOG__SWEEP__PAGE_SIZE=128
 GOLEM__PUBLIC_WORKER_API__CONNECT_TIMEOUT="10s"
 GOLEM__PUBLIC_WORKER_API__HOST="localhost"
 GOLEM__PUBLIC_WORKER_API__HTTP2_KEEP_ALIVE_INTERVAL="10s"

golem-debugging-service/config/debug-worker-executor.toml

@@ -143,6 +143,17 @@ type = "EveryNInvocation"
 [oplog.oplog_processor_snapshotting.config]
 count = 10
 
+[oplog.sweep]
+enabled = true
+interval = "1m"
+max_archives_per_tick = 256
+max_backoff_intervals = 8
+max_concurrency = 4
+max_scanned_per_tick = 4096
+max_tick_duration = "30s"
+max_tracked_agents = 100000
+page_size = 128
+
 [public_worker_api]
 connect_timeout = "10s"
 host = "localhost"
@@ -435,6 +446,17 @@ without_time = false
 # [oplog.oplog_processor_snapshotting.config]
 # count = 10
 # 
+# [oplog.sweep]
+# enabled = true
+# interval = "1m"
+# max_archives_per_tick = 256
+# max_backoff_intervals = 8
+# max_concurrency = 4
+# max_scanned_per_tick = 4096
+# max_tick_duration = "30s"
+# max_tracked_agents = 100000
+# page_size = 128
+# 
 # [public_worker_api]
 # connect_timeout = "10s"
 # host = "localhost"

golem-worker-executor/config/worker-executor.sample.env

@@ -112,6 +112,15 @@ GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MIN_DELAY="100ms"
 GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MULTIPLIER=3.0
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__TYPE="EveryNInvocation"
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__CONFIG__COUNT=10
+GOLEM__OPLOG__SWEEP__ENABLED=true
+GOLEM__OPLOG__SWEEP__INTERVAL="1m"
+GOLEM__OPLOG__SWEEP__MAX_ARCHIVES_PER_TICK=256
+GOLEM__OPLOG__SWEEP__MAX_BACKOFF_INTERVALS=8
+GOLEM__OPLOG__SWEEP__MAX_CONCURRENCY=4
+GOLEM__OPLOG__SWEEP__MAX_SCANNED_PER_TICK=4096
+GOLEM__OPLOG__SWEEP__MAX_TICK_DURATION="30s"
+GOLEM__OPLOG__SWEEP__MAX_TRACKED_AGENTS=100000
+GOLEM__OPLOG__SWEEP__PAGE_SIZE=128
 GOLEM__PUBLIC_WORKER_API__CONNECT_TIMEOUT="10s"
 GOLEM__PUBLIC_WORKER_API__HOST="localhost"
 GOLEM__PUBLIC_WORKER_API__HTTP2_KEEP_ALIVE_INTERVAL="10s"
@@ -360,6 +369,15 @@ GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MIN_DELAY="100ms"
 GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MULTIPLIER=3.0
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__TYPE="EveryNInvocation"
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__CONFIG__COUNT=10
+GOLEM__OPLOG__SWEEP__ENABLED=true
+GOLEM__OPLOG__SWEEP__INTERVAL="1m"
+GOLEM__OPLOG__SWEEP__MAX_ARCHIVES_PER_TICK=256
+GOLEM__OPLOG__SWEEP__MAX_BACKOFF_INTERVALS=8
+GOLEM__OPLOG__SWEEP__MAX_CONCURRENCY=4
+GOLEM__OPLOG__SWEEP__MAX_SCANNED_PER_TICK=4096
+GOLEM__OPLOG__SWEEP__MAX_TICK_DURATION="30s"
+GOLEM__OPLOG__SWEEP__MAX_TRACKED_AGENTS=100000
+GOLEM__OPLOG__SWEEP__PAGE_SIZE=128
 GOLEM__PUBLIC_WORKER_API__CONNECT_TIMEOUT="10s"
 GOLEM__PUBLIC_WORKER_API__HOST="localhost"
 GOLEM__PUBLIC_WORKER_API__HTTP2_KEEP_ALIVE_INTERVAL="10s"
@@ -578,6 +596,15 @@ GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MIN_DELAY="100ms"
 GOLEM__OPLOG__INDEXED_STORAGE_RETRY__MULTIPLIER=3.0
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__TYPE="EveryNInvocation"
 GOLEM__OPLOG__OPLOG_PROCESSOR_SNAPSHOTTING__CONFIG__COUNT=10
+GOLEM__OPLOG__SWEEP__ENABLED=true
+GOLEM__OPLOG__SWEEP__INTERVAL="1m"
+GOLEM__OPLOG__SWEEP__MAX_ARCHIVES_PER_TICK=256
+GOLEM__OPLOG__SWEEP__MAX_BACKOFF_INTERVALS=8
+GOLEM__OPLOG__SWEEP__MAX_CONCURRENCY=4
+GOLEM__OPLOG__SWEEP__MAX_SCANNED_PER_TICK=4096
+GOLEM__OPLOG__SWEEP__MAX_TICK_DURATION="30s"
+GOLEM__OPLOG__SWEEP__MAX_TRACKED_AGENTS=100000
+GOLEM__OPLOG__SWEEP__PAGE_SIZE=128
 GOLEM__PUBLIC_WORKER_API__CONNECT_TIMEOUT="10s"
 GOLEM__PUBLIC_WORKER_API__HOST="localhost"
 GOLEM__PUBLIC_WORKER_API__HTTP2_KEEP_ALIVE_INTERVAL="10s"

golem-worker-executor/config/worker-executor.toml

@@ -182,6 +182,17 @@ type = "EveryNInvocation"
 [oplog.oplog_processor_snapshotting.config]
 count = 10
 
+[oplog.sweep]
+enabled = true
+interval = "1m"
+max_archives_per_tick = 256
+max_backoff_intervals = 8
+max_concurrency = 4
+max_scanned_per_tick = 4096
+max_tick_duration = "30s"
+max_tracked_agents = 100000
+page_size = 128
+
 [public_worker_api]
 connect_timeout = "10s"
 host = "localhost"
@@ -550,6 +561,17 @@ without_time = false
 # [oplog.oplog_processor_snapshotting.config]
 # count = 10
 # 
+# [oplog.sweep]
+# enabled = true
+# interval = "1m"
+# max_archives_per_tick = 256
+# max_backoff_intervals = 8
+# max_concurrency = 4
+# max_scanned_per_tick = 4096
+# max_tick_duration = "30s"
+# max_tracked_agents = 100000
+# page_size = 128
+# 
 # [public_worker_api]
 # connect_timeout = "10s"
 # host = "localhost"
@@ -888,6 +910,17 @@ without_time = false
 # [oplog.oplog_processor_snapshotting.config]
 # count = 10
 # 
+# [oplog.sweep]
+# enabled = true
+# interval = "1m"
+# max_archives_per_tick = 256
+# max_backoff_intervals = 8
+# max_concurrency = 4
+# max_scanned_per_tick = 4096
+# max_tick_duration = "30s"
+# max_tracked_agents = 100000
+# page_size = 128
+# 
 # [public_worker_api]
 # connect_timeout = "10s"
 # host = "localhost"

golem-worker-executor/src/lib.rs

@@ -66,6 +66,7 @@ use crate::services::oplog::{
     BlobOplogArchiveService, CompressedOplogArchiveService, MultiLayerOplogService,
     OplogArchiveService, OplogService, PrimaryOplogService,
 };
+use crate::services::oplog_sweep::OplogSweeper;
 use crate::services::promise::{DefaultPromiseService, DefaultPromiseWorkerAccess, PromiseService};
 use crate::services::quota::QuotaService;
 use crate::services::registry_event_subscriber::WorkerExecutorRegistryInvalidationHandler;
@@ -771,6 +772,9 @@ pub async fn create_worker_executor_impl<
             Arc::new(BlobOplogArchiveService::new(blob_storage.clone(), idx));
         oplog_archives.push(svc);
     }
+    // The sweeper needs the same layer stack, but it is built here while its other dependency
+    // (the worker activator) only exists further down, so keep a handle.
+    let sweep_archives = oplog_archives.clone();
     let oplog_archives = NEVec::try_from_vec(oplog_archives);
 
     let base_oplog_service: Arc<dyn OplogService> = match oplog_archives {
@@ -924,6 +928,25 @@ pub async fn create_worker_executor_impl<
         shutdown_token.clone(),
     );
 
+    // Same work as `ScheduledAction::ArchiveOplog`, driven by a paginated scan of the oplog layers
+    // instead of a row written on the oplog commit path. Spawned into the executor's join set so a
+    // shutdown waits for the tick in flight rather than cutting an archive step in half.
+    let oplog_sweeper = OplogSweeper::over_layers(
+        golem_config.oplog.sweep.clone(),
+        indexed_storage.clone(),
+        &sweep_archives,
+        shard_service.clone(),
+        component_service.clone(),
+        Arc::new(lazy_worker_activator.clone() as Arc<dyn WorkerActivator<Ctx>>),
+    );
+    join_set.spawn({
+        let shutdown_token = shutdown_token.clone();
+        async move {
+            oplog_sweeper.run(shutdown_token).await;
+            Ok(())
+        }
+    });
+
     let additional_deps = bootstrap.create_additional_deps(registry_service.clone());
 
     let direct_invocation_auth_service =

golem-worker-executor/src/metrics.rs

@@ -86,6 +86,12 @@ const SCHEDULER_TICK_DURATION_BUCKETS: &[f64; 24] = &[
     2.5, 3.0, 5.0, 10.0, 15.0, 20.0, 30.0, 60.0,
 ];
 
+/// Tick-duration buckets for the oplog sweep. They reach well past the shared set's minute: a
+/// tick is a budgeted walk of a storage namespace, so slow storage is exactly when its duration
+/// is worth reading, and the shared buckets saturate there.
+const OPLOG_SWEEP_TICK_BUCKETS: &[f64; 9] =
+    &[0.01, 0.1, 1.0, 5.0, 15.0, 60.0, 300.0, 900.0, 3600.0];
+
 /// Buckets for the size of a single `memory.grow` allocation. Deliberately
 /// fine-grained in the 1-32 MiB band where typical guest grows cluster, so
 /// that p90/p99 quantiles are not pinned to a coarse 4-16 MiB bucket edge.
@@ -1009,6 +1015,26 @@ pub mod oplog {
             &["account_id", "environment_id"]
         )
         .unwrap();
+        static ref OPLOG_SWEEP_OUTCOME_TOTAL: CounterVec = register_counter_vec!(
+            "oplog_sweep_outcome_total",
+            "Keys the oplog sweep examined, by what it decided about each. One outcome per key, \
+             so the outcomes sum to what a tick decided",
+            &["route", "outcome"]
+        )
+        .unwrap();
+        static ref OPLOG_SWEEP_TICK_TIME: HistogramVec = register_histogram_vec!(
+            "oplog_sweep_tick_seconds",
+            "Time taken by one oplog sweep tick",
+            &["route"],
+            crate::metrics::OPLOG_SWEEP_TICK_BUCKETS.to_vec()
+        )
+        .unwrap();
+        static ref OPLOG_SWEEP_TRUNCATED_TOTAL: CounterVec = register_counter_vec!(
+            "oplog_sweep_truncated_total",
+            "Oplog sweep ticks that hit a budget or their deadline before reaching the end of the namespace",
+            &["route"]
+        )
+        .unwrap();
         static ref OPLOG_STORAGE_RETRY_TOTAL: CounterVec = register_counter_vec!(
             "oplog_storage_retry_total",
             "Number of oplog storage operation retries due to transient errors",
@@ -1033,6 +1059,25 @@ pub mod oplog {
             .inc();
     }
 
+    pub fn record_oplog_sweep_outcome(route: &str, outcome: &'static str, count: u64) {
+        if count > 0 {
+            OPLOG_SWEEP_OUTCOME_TOTAL
+                .with_label_values(&[route, outcome])
+                .inc_by(count as f64);
+        }
+    }
+
+    pub fn record_oplog_sweep_tick(route: &str, duration: std::time::Duration, truncated: bool) {
+        OPLOG_SWEEP_TICK_TIME
+            .with_label_values(&[route])
+            .observe(duration.as_secs_f64());
+        if truncated {
+            OPLOG_SWEEP_TRUNCATED_TOTAL
+                .with_label_values(&[route])
+                .inc();
+        }
+    }
+
     pub fn record_scheduled_archive(duration: std::time::Duration, has_more: bool) {
         SCHEDULED_ARCHIVE_TIME
             .with_label_values(if has_more {