feat(runtime): add memory observability controller pilot (#3405)

docs/architecture/background-controller-contract.md

@@ -114,6 +114,17 @@ This snapshot intentionally does not define an admin route, scheduler, service
 registry, worker start/stop path, readiness signal, peer signal, storage write,
 or metrics emission change.
 
+## BGC-004 Controller Pilot
+
+The first controller pilot is also memory observability. It converts the
+existing desired inputs and status snapshot into a typed reconcile plan. The
+pilot reports desired state, current state, and worker mutation intent.
+
+The only allowed worker mutation for this pilot is `none`. Repeated reconcile
+calls must return the same plan for the same snapshot and must not request a
+worker start, stop, resize, wakeup, storage write, readiness signal, peer
+signal, or metrics emission.
+
 ## Future Reconcile Rules
 
 Future reconcile work is allowed only after a read-only status snapshot exists.

docs/architecture/background-services-inventory.md

@@ -48,7 +48,7 @@ not define a new scheduler, controller framework, or shutdown contract.
 | Audit runtime | `rustfs/src/server/audit.rs::start_audit_system` starts audit only when module switches and configured targets allow it. `AuditSystem::reload_config` replaces runtime targets. | Dispatches audit events to configured targets and manages replay workers. | Audit observability records config reloads and target delivery metrics. | Do not couple audit lifecycle to notification lifecycle even though the runtime patterns are similar. |
 | Dynamic config reload | Admin config handlers call `apply_dynamic_config_for_subsystem`, then `signal_dynamic_config_reload` or `signal_config_snapshot_reload` through the global notification system. | Applies scanner/heal runtime config, audit reloads, notification reloads, and peer reload signals. | Logs local and peer reload failures. Audit reload increments audit config reload metrics. | This is admin-triggered fanout, not a background scheduler. Controller work should preserve per-subsystem validation and error boundaries. |
 | Metrics runtime | `crates/obs/src/metrics/scheduler.rs::init_metrics_runtime` spawns multiple interval loops for cluster, bucket, node, resource, audit, notification, and replication bandwidth metrics. | Periodically collects and reports metrics. | Reports through the metrics runtime and logs cancellation warnings. | Keep intervals and collector grouping stable while adding status snapshots. |
-| Memory observability | `rustfs/src/memory_observability.rs::init_memory_observability` spawns a token-cancelled sampler. | Periodically records memory snapshots. | Emits memory observability metrics and exposes a read-only status snapshot for metrics enablement, interval, cancellation source, and shutdown handle shape. | This is the first low-risk pilot for read-only controller status because it already has a simple token loop. |
+| Memory observability | `rustfs/src/memory_observability.rs::init_memory_observability` spawns a token-cancelled sampler. | Periodically records memory snapshots. | Emits memory observability metrics and exposes a read-only status snapshot plus a no-op reconcile plan for metrics enablement, interval, cancellation source, and shutdown handle shape. | This is the first low-risk pilot for controller status because it already has a simple token loop and the pilot does not mutate workers. |
 | Allocator reclaim | `rustfs/src/allocator_reclaim.rs::init_allocator_reclaim` spawns a token-cancelled reclaim loop when enabled. | Observes reclaimable work and may run allocator reclaim after idle intervals. | Emits reclaim enabled/backend counters, active-request gauges, scanner/heal activity gauges, and reclaim result counters. | A controller must preserve idle-streak logic and backend-specific force behavior. |
 | Auto-tuner | `rustfs/src/init.rs::init_auto_tuner` optionally spawns a 60-second loop when `RUSTFS_AUTOTUNER_ENABLED` is true. | Tunes concurrency manager settings from performance metrics. | Logs iteration success/failure. | Treat as behavior-sensitive; a future controller needs explicit rollback because it can change runtime concurrency. |
 | Update check | `rustfs/src/init.rs::init_update_check` spawns one async task with a 30-second timeout when update checks are enabled. | Performs version check network I/O and logs available updates. | Logs result only. | This is a one-shot task, not a controller candidate for the first BGC PRs. |

docs/architecture/migration-progress.md

@@ -5,15 +5,16 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 ## Current Context
 
 - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
-- Branch: `overtrue/arch-background-status-snapshot`
-- Baseline: `origin/main` at `f7bf4fd3e4b304c75c880f28d8c45329366571f6`
-- PR type for this branch: `contract`
-- Runtime behavior changes: none; memory observability startup, sampler loop,
-  cancellation, and metrics emission behavior are unchanged.
-- Rust code changes: add a typed read-only memory observability status snapshot
-  and reuse the existing sampler interval parsing helper.
+- Branch: `overtrue/arch-background-controller-pilot`
+- Baseline: `origin/main` at `bda0b1f3dd923f728e4d2ff2f17bde2755a6cb5e`
+- PR type for this branch: `behavior-change`
+- Runtime behavior changes: none; the memory observability controller pilot only
+  returns read-only snapshot/reconcile data and never starts, stops, resizes, or
+  wakes workers.
+- Rust code changes: add a memory observability controller snapshot and
+  reconcile plan with explicit no-op worker mutation.
 - CI/script changes: none.
-- Docs changes: record `BGC-003` status snapshot scope and update background
+- Docs changes: record `BGC-004` controller pilot scope and update background
   controller migration progress.
 
 ## Phase 0 Tasks
@@ -388,26 +389,35 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
     behavior change.
   - Verification: focused memory observability tests, compile checks, migration
     guards, formatting, and pre-commit quality gate.
+- [x] `BGC-004` Pilot one controller.
+  - Acceptance: memory observability exposes a typed controller snapshot and
+    reconcile plan that compare desired state with current status.
+  - Must preserve: no admin route, scheduler, service registry, worker
+    lifecycle mutation, storage write, readiness signal, peer signal, or metrics
+    emission behavior change.
+  - Verification: focused controller tests prove repeated reconcile is
+    idempotent, cancellation state is preserved, and worker mutation remains
+    none.
 
 ## Next PRs
 
-1. `behavior-change`: pilot one low-risk controller behind status-first tests
-   in `BGC-004`.
-2. `test-only`: add controller harness coverage for cancellation and
+1. `test-only`: add controller harness coverage for cancellation and
    duplicate-worker prevention in `TEST-BGC-001`.
+2. `behavior-change`: add the next low-risk background status surface before
+   broader reconcile work.
 
 ## Pre-Push Review Log
 
 | Expert | Status | Notes |
 |---|---|---|
-| Quality/architecture | pass | Confirmed the snapshot is a typed read-only contract in `memory_observability` with no generic controller framework, service registry, or admin route. |
-| Migration preservation | pass | Confirmed memory observability startup, sampler loop, runtime token cancellation, interval defaulting, and metrics emission behavior remain unchanged. |
-| Testing/verification | pass | Confirmed focused tests, rustfs library check, formatting, diff hygiene, migration guards, Rust risk scan, and `make pre-commit` all passed. |
+| Quality/architecture | pass | Confirmed the controller pilot stays local to `memory_observability`, uses typed desired/status/reconcile structs, and does not add a generic scheduler or admin route. |
+| Migration preservation | pass | Confirmed reconcile reports `worker_mutation: none`, preserving sampler startup, cancellation, interval defaulting, and metrics emission behavior. |
+| Testing/verification | pass | Focused tests, compile checks, formatting, diff hygiene, migration guards, Rust risk scan, and `make pre-commit` passed. |
 
 ## Verification Notes
 
-Passed on `f7bf4fd3e1b3b48b3bec0702a6d8a840b45a5160`:
-- `cargo test -p rustfs memory_observability --lib`; 5 passed.
+Passed on `bda0b1f3dd923f728e4d2ff2f17bde2755a6cb5e`:
+- `cargo test -p rustfs memory_observability --lib`; 8 passed.
 - `cargo check -p rustfs --lib`.
 - `cargo fmt --all`.
 - `cargo fmt --all --check`.
@@ -416,20 +426,19 @@ Passed on `f7bf4fd3e1b3b48b3bec0702a6d8a840b45a5160`:
 - `./scripts/check_layer_dependencies.sh`.
 - `./scripts/check_metrics_migration_refs.sh`.
 - Rust risk scan for unwrap/expect/casts/string errors/debug output in
-  `rustfs/src/memory_observability.rs`; only pre-existing test `expect` calls
-  were matched.
-- `make pre-commit`; all checks passed, including nextest 5857 passed and 111
+  `rustfs/src/memory_observability.rs`; only test `expect` calls matched.
+- `make pre-commit`; all checks passed, including nextest 5865 passed and 111
   skipped.
 
 Notes:
-- This slice only exposes in-memory status from existing inputs. It does not
-  start, stop, resize, or wake memory observability workers.
-- There is no admin/API exposure in this PR; future `BGC-004` work must remain
-  status-first before adding reconcile behavior.
+- This slice adds reconcile data only. It does not apply reconcile output to the
+  running sampler.
+- There is no admin/API exposure in this PR; future controller harness work
+  should stay isolated from scanner, heal, lifecycle, and replication.
 
 ## Handoff Notes
 
-- Next migration slice can start `BGC-004` by choosing one low-risk controller
-  and proving idempotent reconcile without duplicate workers.
+- Next migration slice can start `TEST-BGC-001` with a fake-clock and
+  cancellation-token harness around the memory observability pilot.
 - Keep scanner, heal, lifecycle, replication, disk health, and config reload out
   of broad controller movement until dedicated preservation tests exist.

rustfs/src/memory_observability.rs

@@ -61,6 +61,62 @@ pub struct MemoryObservabilityStatusSnapshot {
     pub shutdown_handle: MemoryObservabilityShutdownHandle,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MemoryObservabilityDesiredState {
+    Disabled,
+    Enabled,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+pub struct MemoryObservabilityDesiredSnapshot {
+    pub state: MemoryObservabilityDesiredState,
+    pub interval_secs: u64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+pub struct MemoryObservabilityControllerSnapshot {
+    pub desired: MemoryObservabilityDesiredSnapshot,
+    pub status: MemoryObservabilityStatusSnapshot,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MemoryObservabilityWorkerMutation {
+    None,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+pub struct MemoryObservabilityReconcilePlan {
+    pub service: &'static str,
+    pub desired: MemoryObservabilityDesiredSnapshot,
+    pub current_state: MemoryObservabilityServiceState,
+    pub worker_mutation: MemoryObservabilityWorkerMutation,
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+pub struct MemoryObservabilityController;
+
+impl MemoryObservabilityController {
+    pub fn snapshot(&self, ctx: &CancellationToken) -> MemoryObservabilityControllerSnapshot {
+        memory_observability_controller_snapshot(ctx)
+    }
+
+    pub fn reconcile(&self, ctx: &CancellationToken) -> MemoryObservabilityReconcilePlan {
+        let snapshot = self.snapshot(ctx);
+        self.reconcile_snapshot(snapshot)
+    }
+
+    pub fn reconcile_snapshot(&self, snapshot: MemoryObservabilityControllerSnapshot) -> MemoryObservabilityReconcilePlan {
+        MemoryObservabilityReconcilePlan {
+            service: MEMORY_OBSERVABILITY_SERVICE_NAME,
+            desired: snapshot.desired,
+            current_state: snapshot.status.state,
+            worker_mutation: MemoryObservabilityWorkerMutation::None,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 struct CgroupMemorySnapshot {
     current_bytes: Option<u64>,
@@ -152,6 +208,19 @@ fn configured_memory_observability_interval_secs() -> u64 {
     rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
 }
 
+fn build_memory_observability_desired_snapshot(metrics_enabled: bool, interval_secs: u64) -> MemoryObservabilityDesiredSnapshot {
+    let state = if metrics_enabled {
+        MemoryObservabilityDesiredState::Enabled
+    } else {
+        MemoryObservabilityDesiredState::Disabled
+    };
+
+    MemoryObservabilityDesiredSnapshot {
+        state,
+        interval_secs: interval_secs.max(1),
+    }
+}
+
 fn build_memory_observability_status_snapshot(
     metrics_enabled: bool,
     interval_secs: u64,
@@ -183,6 +252,25 @@ pub fn memory_observability_status_snapshot(ctx: &CancellationToken) -> MemoryOb
     )
 }
 
+fn build_memory_observability_controller_snapshot(
+    metrics_enabled: bool,
+    interval_secs: u64,
+    cancellation_requested: bool,
+) -> MemoryObservabilityControllerSnapshot {
+    MemoryObservabilityControllerSnapshot {
+        desired: build_memory_observability_desired_snapshot(metrics_enabled, interval_secs),
+        status: build_memory_observability_status_snapshot(metrics_enabled, interval_secs, cancellation_requested),
+    }
+}
+
+pub fn memory_observability_controller_snapshot(ctx: &CancellationToken) -> MemoryObservabilityControllerSnapshot {
+    build_memory_observability_controller_snapshot(
+        rustfs_obs::observability_metric_enabled(),
+        configured_memory_observability_interval_secs(),
+        ctx.is_cancelled(),
+    )
+}
+
 async fn record_memory_snapshot() {
     match tokio::task::spawn_blocking(|| {
         let (resource, process) = snapshot_process_resource_and_system();
@@ -239,8 +327,10 @@ pub fn init_memory_observability(ctx: CancellationToken) {
 #[cfg(test)]
 mod tests {
     use super::{
-        CgroupMemorySnapshot, MemoryObservabilityCancellationSource, MemoryObservabilityServiceState,
-        MemoryObservabilityShutdownHandle, build_memory_observability_status_snapshot, parse_kv_stats, read_optional_u64,
+        CgroupMemorySnapshot, MemoryObservabilityCancellationSource, MemoryObservabilityController,
+        MemoryObservabilityDesiredState, MemoryObservabilityServiceState, MemoryObservabilityShutdownHandle,
+        MemoryObservabilityWorkerMutation, build_memory_observability_controller_snapshot,
+        build_memory_observability_status_snapshot, parse_kv_stats, read_optional_u64,
     };
     use std::fs;
     use std::path::PathBuf;
@@ -298,4 +388,41 @@ mod tests {
         assert_eq!(stopping.state, MemoryObservabilityServiceState::Stopping);
         assert_eq!(stopping.interval_secs, 30);
     }
+
+    #[test]
+    fn memory_observability_controller_reconcile_is_idempotent() {
+        let controller = MemoryObservabilityController;
+        let snapshot = build_memory_observability_controller_snapshot(true, 15, false);
+
+        let first = controller.reconcile_snapshot(snapshot);
+        let second = controller.reconcile_snapshot(snapshot);
+
+        assert_eq!(first, second);
+        assert_eq!(first.desired.state, MemoryObservabilityDesiredState::Enabled);
+        assert_eq!(first.current_state, MemoryObservabilityServiceState::Running);
+        assert_eq!(first.worker_mutation, MemoryObservabilityWorkerMutation::None);
+    }
+
+    #[test]
+    fn memory_observability_controller_preserves_cancellation_state_without_worker_mutation() {
+        let controller = MemoryObservabilityController;
+        let snapshot = build_memory_observability_controller_snapshot(true, 30, true);
+        let plan = controller.reconcile_snapshot(snapshot);
+
+        assert_eq!(snapshot.status.state, MemoryObservabilityServiceState::Stopping);
+        assert_eq!(plan.current_state, MemoryObservabilityServiceState::Stopping);
+        assert_eq!(plan.worker_mutation, MemoryObservabilityWorkerMutation::None);
+    }
+
+    #[test]
+    fn memory_observability_controller_reports_disabled_desired_state_without_starting_worker() {
+        let controller = MemoryObservabilityController;
+        let snapshot = build_memory_observability_controller_snapshot(false, 0, false);
+        let plan = controller.reconcile_snapshot(snapshot);
+
+        assert_eq!(snapshot.desired.state, MemoryObservabilityDesiredState::Disabled);
+        assert_eq!(snapshot.desired.interval_secs, 1);
+        assert_eq!(plan.current_state, MemoryObservabilityServiceState::Disabled);
+        assert_eq!(plan.worker_mutation, MemoryObservabilityWorkerMutation::None);
+    }
 }