feat(ecstore): add the on-demand migration backfill job (#7087)

* feat(ecstore): add on-demand migration backfill job core

Add the background backfill job for on-demand migration
(rustfs/backlog#2159): a durable checkpoint under
buckets/<bucket>/on-demand-migration-backfill.json saved by If-Match
compare-and-set every 1000 keys or 10 s, a 60 s owner lease renewed by
every save, a recovery pass that takes over expired leases (or jobs this
node owned before a restart) and cancels jobs whose config changed, and a
main loop over the source ListObjectsV2 pages with the skip_existing
policy, dry runs, bounded outstanding pulls and wait-on-full enqueueing.

The pull queue gains per-job completion reports so the job can count
pulled/failed keys (hashes only), and pull permits become two-tier so an
online miss is never queued behind a backfill pull.

* feat(admin): expose on-demand migration backfill job

Wire the ODM-12 backfill job (rustfs/backlog#2159) to its operators:
POST /v3/on-demand-migration/{bucket}/backfill?op=start|cancel and
GET .../backfill return the checkpoint document, GET .../status gains a
backfill summary, and the recovery loop plus the process-wide runner are
installed at startup. Backfill control reuses
Set/GetBucketOnDemandMigrationAction and is recorded in the route policy,
the registration matrix and the admin route snapshot.

Add the rustfs-madmin wire types and client methods with golden fixtures
shared by the server tests, the backfill_* metric descriptors and their
collector, and three e2e scenarios: a full backfill across list pages,
cancellation, and resuming from the persisted continuation token after a
server restart.

crates/e2e_test/src/on_demand_migration/backfill_test.rs

@@ -0,0 +1,254 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! On-demand migration backfill job scenarios (ODM-12, rustfs/backlog#2159):
+//! a full backfill of a small-object source, cancellation, and resuming from
+//! the persisted continuation token after a server restart.
+
+use super::common::{BackfillOp, BackfillRequest, ODM_SERVER_ENV, OdmSourceSpec, OdmTestEnv, SeedObject};
+use crate::fake_s3_target::Operation;
+use bytes::Bytes;
+use std::time::Duration;
+
+type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
+
+const SOURCE_BUCKET: &str = "odm-backfill-source";
+const LOCAL_BUCKET: &str = "odm-backfill-local";
+const KEY_PREFIX: &str = "cold/";
+/// Keys per scenario. The fake source retains at most 4,096 object versions
+/// and 4,096 journal entries, and one pull is a HEAD plus a GET, so a
+/// scenario that asserts on the journal stays below ~2,000 keys. The job
+/// lists 1,000 keys per page, so this still spans several pages and exercises
+/// the continuation token, which is what the scenarios are about.
+const SEEDED_KEYS: usize = 1500;
+
+fn key(i: usize) -> String {
+    format!("{KEY_PREFIX}{i:05}")
+}
+
+/// Content that identifies the key so a mis-stored object is caught.
+fn body(i: usize) -> Bytes {
+    Bytes::from(format!("object-{i:05}-payload"))
+}
+
+fn seed(env: &OdmTestEnv, count: usize) {
+    let objects: Vec<SeedObject> = (0..count).map(|i| SeedObject::new(key(i), body(i))).collect();
+    let etags = env.seed_source(SOURCE_BUCKET, &objects);
+    assert_eq!(etags.len(), count);
+}
+
+async fn configure(env: &OdmTestEnv, spec: &OdmSourceSpec) -> TestResult {
+    let response = env.configure_source(LOCAL_BUCKET, spec).await?;
+    assert_eq!(response.status, 200, "configure: {}", response.body);
+    // The probe issued a one-key listing; count only the job's traffic.
+    env.source.take_requests();
+    Ok(())
+}
+
+fn source_lists(env: &OdmTestEnv) -> Vec<Option<String>> {
+    env.source
+        .requests()
+        .into_iter()
+        .filter(|record| record.operation == Operation::ListObjectsV2)
+        .map(|record| record.continuation_token)
+        .collect()
+}
+
+fn source_gets(env: &OdmTestEnv) -> usize {
+    env.source
+        .requests()
+        .into_iter()
+        .filter(|record| record.operation == Operation::GetObject)
+        .count()
+}
+
+fn counter(job: &serde_json::Value, name: &str) -> u64 {
+    job[name].as_u64().unwrap_or_else(|| panic!("{name} missing in {job}"))
+}
+
+#[tokio::test]
+async fn backfill_pulls_every_source_object_across_list_pages() -> TestResult {
+    const COUNT: usize = SEEDED_KEYS;
+    let env = OdmTestEnv::start().await?;
+    env.source.create_bucket(SOURCE_BUCKET);
+    env.rustfs.create_test_bucket(LOCAL_BUCKET).await?;
+    seed(&env, COUNT);
+    configure(&env, &env.fake_source_spec(SOURCE_BUCKET)).await?;
+
+    let started = env.start_backfill(LOCAL_BUCKET, BackfillRequest::default()).await?;
+    assert_eq!(started.status, 200, "start: {}", started.body);
+    let job = started.json()?["job"].clone();
+    assert_eq!(job["state"], "running");
+    assert_eq!(job["skip_existing"], "always");
+    assert_eq!(job["dry_run"], false);
+    let job_id = job["job_id"].as_str().expect("job id").to_string();
+
+    // A second start while the job holds its lease is a conflict.
+    let again = env
+        .backfill(LOCAL_BUCKET, BackfillOp::Start(BackfillRequest::default()))
+        .await?;
+    assert_eq!(again.status, 409, "second start: {}", again.body);
+    assert!(again.body.contains("OnDemandMigrationBackfillRunning"), "{}", again.body);
+
+    let done = env
+        .wait_for_backfill(LOCAL_BUCKET, Duration::from_secs(240), |job| job["state"] == "completed")
+        .await?;
+    assert_eq!(done["job_id"], job_id.as_str());
+    assert_eq!(counter(&done, "listed"), COUNT as u64);
+    assert_eq!(counter(&done, "enqueued"), COUNT as u64);
+    assert_eq!(counter(&done, "pulled"), COUNT as u64);
+    assert_eq!(counter(&done, "failed"), 0);
+    assert_eq!(counter(&done, "skipped_existing"), 0);
+    assert!(done["continuation_token"].is_null(), "a finished job carries no cursor");
+    assert_eq!(done["last_key"], key(COUNT - 1));
+    assert!(done["failed_keys"].as_array().is_some_and(Vec::is_empty));
+    let expected_bytes: u64 = (0..COUNT).map(|i| body(i).len() as u64).sum();
+    assert_eq!(counter(&done, "bytes"), expected_bytes);
+
+    assert_eq!(env.local_key_count(LOCAL_BUCKET, KEY_PREFIX).await?, COUNT);
+    for i in [0, 999, 1000, 1200, COUNT - 1] {
+        env.assert_local_present(LOCAL_BUCKET, &key(i), &body(i)).await;
+    }
+
+    let lists = source_lists(&env);
+    assert_eq!(lists.len(), COUNT.div_ceil(1000), "{COUNT} keys at 1000 per page: {lists:?}");
+    assert!(lists[0].is_none(), "the first page starts without a cursor");
+    assert!(lists[1..].iter().all(Option::is_some), "every later page carries the cursor");
+    assert_eq!(source_gets(&env), COUNT, "every object is fetched exactly once");
+
+    // The status endpoint summarises the same job.
+    let status = env.status(LOCAL_BUCKET).await?;
+    assert_eq!(status.status, 200);
+    let summary = status.json()?["backfill"].clone();
+    assert_eq!(summary["job_id"], job_id.as_str());
+    assert_eq!(summary["state"], "completed");
+    assert_eq!(counter(&summary, "pulled"), COUNT as u64);
+    Ok(())
+}
+
+#[tokio::test]
+async fn backfill_cancel_stops_enqueueing_and_persists_cancelled() -> TestResult {
+    const COUNT: usize = SEEDED_KEYS;
+    let env = OdmTestEnv::start().await?;
+    env.source.create_bucket(SOURCE_BUCKET);
+    env.rustfs.create_test_bucket(LOCAL_BUCKET).await?;
+    seed(&env, COUNT);
+    let mut spec = env.fake_source_spec(SOURCE_BUCKET);
+    spec.policy.max_concurrent_pulls = 1;
+    configure(&env, &spec).await?;
+
+    // Cancelling before any job exists is a 404, not a silent success.
+    let nothing = env.backfill(LOCAL_BUCKET, BackfillOp::Cancel).await?;
+    assert_eq!(nothing.status, 404, "cancel without a job: {}", nothing.body);
+    assert!(nothing.body.contains("NoSuchBackfillJob"), "{}", nothing.body);
+    let unread = env.backfill(LOCAL_BUCKET, BackfillOp::Status).await?;
+    assert_eq!(unread.status, 404, "status without a job: {}", unread.body);
+
+    let started = env.start_backfill(LOCAL_BUCKET, BackfillRequest::default()).await?;
+    assert_eq!(started.status, 200, "start: {}", started.body);
+    env.wait_for_backfill(LOCAL_BUCKET, Duration::from_secs(60), |job| {
+        job["state"] == "running" && counter(job, "enqueued") > 0
+    })
+    .await?;
+
+    let cancelled = env.backfill(LOCAL_BUCKET, BackfillOp::Cancel).await?;
+    assert_eq!(cancelled.status, 200, "cancel: {}", cancelled.body);
+    let job = cancelled.json()?["job"].clone();
+    assert_eq!(job["state"], "cancelled");
+    let enqueued_at_cancel = counter(&job, "enqueued");
+    assert!(enqueued_at_cancel < COUNT as u64, "the job was cancelled mid-way: {job}");
+
+    // Nothing is queued after the cancel: the checkpoint and the source
+    // traffic both stop moving once the few in-flight pulls drain.
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let persisted = env.backfill_job(LOCAL_BUCKET).await?.expect("checkpoint kept for inspection");
+    assert_eq!(persisted["state"], "cancelled");
+    assert_eq!(counter(&persisted, "enqueued"), enqueued_at_cancel);
+    let gets_after_drain = source_gets(&env);
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    assert_eq!(source_gets(&env), gets_after_drain, "no source GET after the cancel drained");
+    assert!(env.local_key_count(LOCAL_BUCKET, KEY_PREFIX).await? < COUNT);
+
+    // Cancel is idempotent and the status endpoint reports the final state.
+    let again = env.backfill(LOCAL_BUCKET, BackfillOp::Cancel).await?;
+    assert_eq!(again.status, 200, "second cancel: {}", again.body);
+    assert_eq!(again.json()?["job"]["state"], "cancelled");
+    let status = env.status(LOCAL_BUCKET).await?;
+    assert_eq!(status.json()?["backfill"]["state"], "cancelled");
+
+    // A cancelled job releases the bucket: a new job can start.
+    let restarted = env.start_backfill(LOCAL_BUCKET, BackfillRequest::default()).await?;
+    assert_eq!(restarted.status, 200, "restart after cancel: {}", restarted.body);
+    assert_ne!(restarted.json()?["job"]["job_id"], job["job_id"]);
+    let _ = env.backfill(LOCAL_BUCKET, BackfillOp::Cancel).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn backfill_resumes_from_continuation_token_after_restart() -> TestResult {
+    const COUNT: usize = SEEDED_KEYS;
+    let mut env = OdmTestEnv::start().await?;
+    env.source.create_bucket(SOURCE_BUCKET);
+    env.rustfs.create_test_bucket(LOCAL_BUCKET).await?;
+    seed(&env, COUNT);
+    let mut spec = env.fake_source_spec(SOURCE_BUCKET);
+    spec.policy.max_concurrent_pulls = 2;
+    configure(&env, &spec).await?;
+
+    let started = env.start_backfill(LOCAL_BUCKET, BackfillRequest::default()).await?;
+    assert_eq!(started.status, 200, "start: {}", started.body);
+    let job_id = started.json()?["job"]["job_id"].as_str().expect("job id").to_string();
+
+    // Wait for the first page to be committed (cursor persisted), then kill
+    // the server while the job is still running.
+    let mid = env
+        .wait_for_backfill(LOCAL_BUCKET, Duration::from_secs(120), |job| {
+            job["state"] == "running" && job["continuation_token"].is_string()
+        })
+        .await?;
+    assert!(counter(&mid, "listed") >= 1000 && counter(&mid, "listed") < COUNT as u64, "{mid}");
+    env.source.take_requests();
+    env.rustfs.restart_server_preserving_data(vec![], ODM_SERVER_ENV).await?;
+
+    let done = env
+        .wait_for_backfill(LOCAL_BUCKET, Duration::from_secs(300), |job| job["state"] == "completed")
+        .await?;
+    assert_eq!(done["job_id"], job_id.as_str(), "the same job continues after the restart");
+    assert_eq!(counter(&done, "failed"), 0);
+    assert!(
+        counter(&done, "listed") >= COUNT as u64,
+        "the resumed job listed the rest (the interrupted page is listed twice): {done}"
+    );
+    // Keys pulled before the crash are re-listed and skipped, never re-pulled;
+    // a pull whose report died with the old process is counted neither way,
+    // so only the lower bound and the queue accounting are exact.
+    assert!(counter(&done, "pulled") + counter(&done, "skipped_existing") >= COUNT as u64, "{done}");
+    assert!(counter(&done, "pulled") <= counter(&done, "enqueued"), "{done}");
+    assert_eq!(env.local_key_count(LOCAL_BUCKET, KEY_PREFIX).await?, COUNT);
+    for i in [0, 500, 999, 1000, COUNT - 1] {
+        env.assert_local_present(LOCAL_BUCKET, &key(i), &body(i)).await;
+    }
+
+    let lists = source_lists(&env);
+    assert!(!lists.is_empty(), "the resumed job listed the source");
+    assert!(
+        lists.iter().all(Option::is_some),
+        "after the restart every source listing carries a continuation-token: {lists:?}"
+    );
+    assert!(
+        lists.len() <= COUNT.div_ceil(1000),
+        "the listing did not start over from the first page: {lists:?}"
+    );
+    Ok(())
+}

crates/e2e_test/src/on_demand_migration/common.rs

@@ -40,8 +40,12 @@ pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
 /// turns it on so scenario tests exercise the feature without repeating it.
 pub const ODM_MODULE_SWITCH_ENV: &str = "RUSTFS_ON_DEMAND_MIGRATION_ENABLED";
 /// The source client shares the replication egress guard, which rejects the
-/// fake source's loopback endpoint unless this switch is set.
+/// fake source's loopback endpoint unless this switch is set. This is the
+/// documented harness opt-in; see
+/// `docs/operations/outbound-connection-policy.md`.
 pub const ALLOW_LOOPBACK_SOURCE_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET";
+/// Server environment every ODM scenario starts (and restarts) with.
+pub const ODM_SERVER_ENV: &[(&str, &str)] = &[(ODM_MODULE_SWITCH_ENV, "true"), (ALLOW_LOOPBACK_SOURCE_ENV, "true")];
 /// Admin route prefix; the bucket name is appended as one path segment.
 pub const ODM_ADMIN_ROUTE: &str = "/rustfs/admin/v3/on-demand-migration";
 /// Region the fake source is addressed with (it accepts any SigV4 region).
@@ -324,7 +328,7 @@ impl OdmTestEnv {
     pub async fn start_with(options: OdmEnvOptions<'_>) -> Result<Self, BoxError> {
         let source = FakeS3Target::start_with_options(options.source).await?;
         let mut rustfs = RustFSTestEnvironment::new().await?;
-        let mut env: Vec<(&str, &str)> = vec![(ODM_MODULE_SWITCH_ENV, "true"), (ALLOW_LOOPBACK_SOURCE_ENV, "true")];
+        let mut env: Vec<(&str, &str)> = ODM_SERVER_ENV.to_vec();
         for (name, value) in &options.env {
             match env.iter_mut().find(|(existing, _)| existing == name) {
                 Some(entry) => entry.1 = value,
@@ -412,6 +416,81 @@ impl OdmTestEnv {
         }
     }
 
+    /// `POST .../{bucket}/backfill?op=start`, retried while the server
+    /// answers `OnDemandMigrationDisabled`: the bucket state is built
+    /// asynchronously right after the config `PUT`, so an immediate start can
+    /// race it. Any other answer is returned as is.
+    pub async fn start_backfill(&self, bucket: &str, request: BackfillRequest) -> Result<AdminResponse, BoxError> {
+        let deadline = Instant::now() + Duration::from_secs(15);
+        loop {
+            let response = self.backfill(bucket, BackfillOp::Start(request.clone())).await?;
+            let state_not_ready = response.status == 400 && response.body.contains("OnDemandMigrationDisabled");
+            if !state_not_ready || Instant::now() >= deadline {
+                return Ok(response);
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    }
+
+    /// The `job` document of `GET .../{bucket}/backfill`, `None` on 404.
+    pub async fn backfill_job(&self, bucket: &str) -> Result<Option<serde_json::Value>, BoxError> {
+        let response = self.backfill(bucket, BackfillOp::Status).await?;
+        match response.status {
+            200 => Ok(Some(response.json()?["job"].clone())),
+            404 => Ok(None),
+            status => Err(format!("GET backfill answered {status}: {}", response.body).into()),
+        }
+    }
+
+    /// Polls the backfill checkpoint until `accept` returns true or `timeout`
+    /// elapses (an error naming the last observed document).
+    pub async fn wait_for_backfill(
+        &self,
+        bucket: &str,
+        timeout: Duration,
+        accept: impl Fn(&serde_json::Value) -> bool,
+    ) -> Result<serde_json::Value, BoxError> {
+        let deadline = Instant::now() + timeout;
+        let mut last = serde_json::Value::Null;
+        loop {
+            if let Some(job) = self.backfill_job(bucket).await? {
+                if accept(&job) {
+                    return Ok(job);
+                }
+                last = job;
+            }
+            if Instant::now() >= deadline {
+                return Err(
+                    format!("backfill of {bucket} did not reach the expected state within {timeout:?}; last: {last}").into(),
+                );
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    }
+
+    /// Number of keys under `prefix` listed by the RustFS under test (local
+    /// state only, no migration side effects).
+    pub async fn local_key_count(&self, bucket: &str, prefix: &str) -> Result<usize, BoxError> {
+        let mut count = 0;
+        let mut token: Option<String> = None;
+        loop {
+            let page = self
+                .client
+                .list_objects_v2()
+                .bucket(bucket)
+                .prefix(prefix)
+                .max_keys(1000)
+                .set_continuation_token(token.take())
+                .send()
+                .await?;
+            count += page.contents().len();
+            match page.next_continuation_token() {
+                Some(next) if page.is_truncated().unwrap_or(false) => token = Some(next.to_string()),
+                _ => return Ok(count),
+            }
+        }
+    }
+
     async fn admin(
         &self,
         method: http::Method,

crates/e2e_test/src/on_demand_migration/mod.rs

@@ -17,13 +17,15 @@
 //! `common` is the shared environment: one RustFS under test, one programmable
 //! fake S3 source, admin-API wrappers, seeding and local-state assertions.
 //! `harness_self_test` proves the harness itself; `get_basic_test` covers the
-//! GET read-through (rustfs/backlog#2156). The fault, concurrency,
+//! GET read-through (rustfs/backlog#2156) and `backfill_test` the background
+//! backfill job (ODM-12, rustfs/backlog#2159). The fault, concurrency,
 //! interaction and real-source matrix is rustfs/backlog#2158; its lane split
 //! lives in `.config/nextest.toml` (fault / concurrency / real source run
 //! nightly, the rest in the merge lane).
 
 pub mod common;
 
+mod backfill_test;
 mod concurrency_test;
 mod fault_test;
 mod get_basic_test;

crates/ecstore/src/api/mod.rs

@@ -161,9 +161,20 @@ pub mod bucket {
         };
         pub use crate::bucket::on_demand_migration::{
             EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS,
-            PullCompletion, PullQueue, PullReason, PullSource, SourceBody, WriteBackBody, WriteBackError, WriteBackOutcome,
-            WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with,
+            PullCompletion, PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, WriteBackBody, WriteBackError,
+            WriteBackOutcome, WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with,
         };
+        pub mod backfill {
+            pub use crate::bucket::on_demand_migration::backfill::{
+                BACKFILL_CHECKPOINT_FILE, BACKFILL_CHECKPOINT_FORMAT_VERSION, BACKFILL_FAILED_KEYS_CAPACITY, BACKFILL_LEASE,
+                BACKFILL_LEASE_LOCK_PREFIX, BACKFILL_LIST_PAGE_SIZE, BACKFILL_RECOVERY_INTERVAL, BACKFILL_SAVE_EVERY_KEYS,
+                BACKFILL_SAVE_INTERVAL, BackfillCheckpoint, BackfillContext, BackfillContextFactory, BackfillError,
+                BackfillLastError, BackfillOwner, BackfillRecoveryStats, BackfillRequest, BackfillRunner, BackfillState,
+                BucketBackfillContext, LocalBackfillObject, PriorityPullPermits, PullPermit, PullPriority, SkipExisting,
+                StoredCheckpoint, SysBackfillContexts, global_backfill_runner, install_global_backfill_runner, key_hash,
+                read_checkpoint, run_backfill_recovery_loop, spawn_backfill_recovery_loop,
+            };
+        }
         pub mod source_client {
             pub use crate::bucket::on_demand_migration::source_client::{
                 SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,

crates/ecstore/src/bucket/on_demand_migration/mod.rs

@@ -20,6 +20,7 @@
 //! clients guarded by a breaker, a negative cache, singleflight and a pull
 //! concurrency limit (rustfs/backlog#2147).
 
+pub mod backfill;
 pub mod breaker;
 pub mod config;
 pub mod negative_cache;
@@ -40,8 +41,8 @@ pub use config::{
 pub use negative_cache::{NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache};
 pub use pull::{
     EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS, PullCompletion,
-    PullQueue, PullReason, PullSource, SourceBody, WriteBackBody, WriteBackError, WriteBackOutcome, WriteBackPart,
-    WriteBackRequest, commit_inline, commit_inline_with,
+    PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, WriteBackBody, WriteBackError, WriteBackOutcome,
+    WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with,
 };
 pub use stats::{
     GaugeGuard, LastSourceError, LatencyBucketSnapshot, OdmOp, OdmOutcome, OdmStats, OdmStatsSnapshot, PullFailureReason,

crates/ecstore/src/bucket/on_demand_migration/pull.rs

@@ -40,6 +40,7 @@
 //! oversized body therefore fails the local write before it can commit,
 //! independently of the digest check the write-back performs.
 
+use super::backfill::PullPriority;
 use super::source_client::{SourceClient, SourceError, SourceHead};
 use super::stats::{PullFailureReason, PullPath};
 use super::sys::{BucketOdmState, OnDemandMigrationSys, PullError, PullOutcome, PullSlot};
@@ -57,7 +58,7 @@ use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 use time::OffsetDateTime;
 use tokio::sync::mpsc::{self, error::TrySendError};
-use tokio::sync::watch;
+use tokio::sync::{oneshot, watch};
 use tokio::task::JoinSet;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, trace};
@@ -108,6 +109,27 @@ impl PullReason {
             PullReason::Backfill => PullPath::Backfill,
         }
     }
+
+    /// Backfill pulls yield pull permits to online requests.
+    pub fn priority(self) -> PullPriority {
+        match self {
+            PullReason::RangeGet | PullReason::LargeObject => PullPriority::Online,
+            PullReason::Backfill => PullPriority::Backfill,
+        }
+    }
+}
+
+/// How one queued pull ended, delivered to the requester that asked for a
+/// report (the backfill job counts these).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum QueuedPullOutcome {
+    /// A new local object of `size` bytes was written.
+    Stored {
+        size: u64,
+    },
+    /// A current local version already existed; nothing was pulled.
+    AlreadyPresent,
+    Failed(PullError),
 }
 
 /// Result of [`PullQueue::enqueue`].
@@ -731,6 +753,8 @@ pub async fn commit_inline_with(
 struct PullJob {
     key: String,
     reason: PullReason,
+    /// Dropped without a send when the job is cancelled before it runs.
+    report: Option<oneshot::Sender<QueuedPullOutcome>>,
 }
 
 /// Bounded per-bucket queue of background pulls. Keys are unique while
@@ -809,26 +833,39 @@ impl PullQueue {
     }
 
     pub fn enqueue(&self, key: &str, reason: PullReason) -> EnqueueOutcome {
+        self.enqueue_with_report(key, reason).0
+    }
+
+    /// [`Self::enqueue`] that also hands back the job's report channel when
+    /// a new job was queued (`Coalesced` pulls report to their first
+    /// requester only).
+    pub fn enqueue_with_report(
+        &self,
+        key: &str,
+        reason: PullReason,
+    ) -> (EnqueueOutcome, Option<oneshot::Receiver<QueuedPullOutcome>>) {
         if self.cancel.is_cancelled() {
-            return EnqueueOutcome::Unavailable;
+            return (EnqueueOutcome::Unavailable, None);
         }
         let mut pending = self.pending.lock();
         if pending.contains(key) {
-            return EnqueueOutcome::Coalesced;
+            return (EnqueueOutcome::Coalesced, None);
         }
+        let (report_tx, report_rx) = oneshot::channel();
         match self.tx.try_send(PullJob {
             key: key.to_string(),
             reason,
+            report: Some(report_tx),
         }) {
             Ok(()) => {
                 pending.insert(key.to_string());
-                EnqueueOutcome::Enqueued
+                (EnqueueOutcome::Enqueued, Some(report_rx))
             }
             Err(TrySendError::Full(_)) => {
                 self.stats.record_pull_failure(PullFailureReason::QueueFull);
-                EnqueueOutcome::QueueFull
+                (EnqueueOutcome::QueueFull, None)
             }
-            Err(TrySendError::Closed(_)) => EnqueueOutcome::Unavailable,
+            Err(TrySendError::Closed(_)) => (EnqueueOutcome::Unavailable, None),
         }
     }
 }
@@ -860,7 +897,7 @@ async fn dispatch(
             queue: Arc::clone(&queue),
             key: job.key.clone(),
         };
-        let slot = match state.acquire_pull_slot(&job.key).await {
+        let slot = match state.acquire_pull_slot_with_priority(&job.key, job.reason.priority()).await {
             Ok(slot) => slot,
             Err(err) => {
                 let result: Result<PullCompletion, PullError> = Err(err);
@@ -905,11 +942,14 @@ async fn run_job(
 ) {
     let _pending = pending;
     let cancel = state.cancel_token();
-    match slot {
+    let report = match slot {
         PullSlot::Follower(follower) => {
             // Someone else (inline GET or an earlier job) is pulling the key;
             // its result makes this job redundant.
-            let _ = follower.wait().await;
+            match follower.wait().await {
+                Ok(outcome) => QueuedPullOutcome::Stored { size: outcome.size },
+                Err(err) => QueuedPullOutcome::Failed(err),
+            }
         }
         PullSlot::Leader(leader) => {
             let ctx = PullContext {
@@ -921,8 +961,17 @@ async fn run_job(
             };
             let result = pull_object(&ctx).await;
             record_completion(&state, &job.key, job.reason.path(), &result);
+            let report = match &result {
+                Ok(PullCompletion::Stored(outcome)) => QueuedPullOutcome::Stored { size: outcome.size },
+                Ok(PullCompletion::AlreadyPresent(_)) => QueuedPullOutcome::AlreadyPresent,
+                Err(err) => QueuedPullOutcome::Failed(err.clone()),
+            };
             leader.complete(result.map(|completion| completion.outcome()));
+            report
         }
+    };
+    if let Some(tx) = job.report {
+        let _ = tx.send(report);
     }
 }
 
@@ -948,9 +997,18 @@ impl BucketOdmState {
 
     /// Queues a background pull of `key`; see [`EnqueueOutcome`].
     pub fn enqueue_pull(self: &Arc<Self>, key: &str, reason: PullReason) -> EnqueueOutcome {
+        self.enqueue_pull_with_report(key, reason).0
+    }
+
+    /// [`Self::enqueue_pull`] with the job's report channel.
+    pub fn enqueue_pull_with_report(
+        self: &Arc<Self>,
+        key: &str,
+        reason: PullReason,
+    ) -> (EnqueueOutcome, Option<oneshot::Receiver<QueuedPullOutcome>>) {
         match self.pull_queue() {
-            Some(queue) => queue.enqueue(key, reason),
-            None => EnqueueOutcome::Unavailable,
+            Some(queue) => queue.enqueue_with_report(key, reason),
+            None => (EnqueueOutcome::Unavailable, None),
         }
     }
 }

crates/ecstore/src/bucket/on_demand_migration/sys.rs

@@ -39,6 +39,7 @@
 //! (`pull.rs`) stores objects with; each bucket state captures it at build
 //! time together with its lazily started [`PullQueue`].
 
+use super::backfill::{PriorityPullPermits, PullPermit, PullPriority};
 use super::breaker::{Breaker, BreakerState, BreakerTransition, BreakerVerdict};
 use super::config::{
     ON_DEMAND_MIGRATION_CONFIG_HOOK, OnDemandMigrationConfig, PathStyle as ConfigPathStyle, Provider, SourceConfig,
@@ -57,7 +58,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 use time::OffsetDateTime;
-use tokio::sync::{OwnedSemaphorePermit, Semaphore, watch};
+use tokio::sync::watch;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
@@ -182,7 +183,7 @@ pub struct PullLeader {
     state: Arc<BucketOdmState>,
     key: String,
     tx: watch::Sender<Option<PullResult>>,
-    _permit: OwnedSemaphorePermit,
+    _permit: PullPermit,
     _inflight: GaugeGuard,
     completed: bool,
 }
@@ -269,7 +270,8 @@ pub struct BucketOdmState {
     breaker: Breaker,
     negative_cache: NegativeCache,
     inflight: Mutex<HashMap<String, watch::Receiver<Option<PullResult>>>>,
-    pull_semaphore: Arc<Semaphore>,
+    /// Online misses first, backfill pulls when nobody waits (ODM-12).
+    pull_permits: Arc<PriorityPullPermits>,
     stats: Arc<OdmStats>,
     cancel: CancellationToken,
     last_source_error_logged_at: Mutex<Option<Instant>>,
@@ -318,7 +320,7 @@ impl BucketOdmState {
             breaker: Breaker::new(),
             negative_cache: NegativeCache::new(Duration::from_secs(policy.negative_cache_ttl_secs)),
             inflight: Mutex::new(HashMap::new()),
-            pull_semaphore: Arc::new(Semaphore::new(policy.max_concurrent_pulls.max(1) as usize)),
+            pull_permits: PriorityPullPermits::new(policy.max_concurrent_pulls.max(1) as usize),
             stats,
             cancel: CancellationToken::new(),
             last_source_error_logged_at: Mutex::new(None),
@@ -477,6 +479,16 @@ impl BucketOdmState {
     /// later callers for the same key become followers and never touch the
     /// semaphore. Fails with `Canceled` when the state is torn down.
     pub async fn acquire_pull_slot(self: &Arc<Self>, key: &str) -> Result<PullSlot, PullError> {
+        self.acquire_pull_slot_with_priority(key, PullPriority::Online).await
+    }
+
+    /// [`Self::acquire_pull_slot`] at the given permit priority; the pull
+    /// queue passes `Backfill` for backfill jobs.
+    pub async fn acquire_pull_slot_with_priority(
+        self: &Arc<Self>,
+        key: &str,
+        priority: PullPriority,
+    ) -> Result<PullSlot, PullError> {
         if self.cancel.is_cancelled() {
             return Err(PullError::canceled("bucket on-demand migration state was removed"));
         }
@@ -499,7 +511,7 @@ impl BucketOdmState {
         let permit = {
             let _queued = self.stats.queue_guard();
             tokio::select! {
-                permit = Arc::clone(&self.pull_semaphore).acquire_owned() => permit,
+                permit = self.pull_permits.acquire(priority) => permit,
                 _ = self.cancel.cancelled() => {
                     return Err(PullError::canceled("bucket on-demand migration state was removed"));
                 }
@@ -524,6 +536,10 @@ impl BucketOdmState {
         self.inflight.lock().len()
     }
 
+    pub fn pull_permits(&self) -> &Arc<PriorityPullPermits> {
+        &self.pull_permits
+    }
+
     pub fn snapshot(&self) -> OdmBucketSnapshot {
         OdmBucketSnapshot {
             bucket: self.bucket.clone(),

crates/madmin/fixtures/on_demand_migration/backfill_job.json

@@ -0,0 +1 @@
+{"bucket":"photos","job":{"format_version":1,"job_id":"11111111-1111-4111-8111-111111111111","state":"running","config_updated_at":"2026-09-02T10:00:00Z","prefix":"photos/","skip_existing":"always","dry_run":false,"continuation_token":"cGhvdG9zLzEwMDA=","listed":2000,"enqueued":1500,"pulled":1400,"skipped_existing":500,"failed":3,"bytes":73400320,"last_key":"photos/2024/02.jpg","last_error":{"class":"source_timeout","key_hash":"9f2c3b0a1d4e5f60","at":"2026-09-02T10:05:00Z"},"failed_keys":["9f2c3b0a1d4e5f60"],"started_at":"2026-09-02T10:00:30Z","updated_at":"2026-09-02T10:05:10Z","owner":{"node":"node-a:9000","lease_until":"2026-09-02T10:06:10Z"}}}

crates/madmin/fixtures/on_demand_migration/status_with_backfill.json

@@ -0,0 +1 @@
+{"configured":true,"enabled":true,"module_enabled":true,"provider":"minio","endpoint_host":"source.example.com","breaker":{"state":"half_open","opened_at":null},"counters":{"requests_total":{"get":{"breaker_open":0,"filtered":0,"negative_cached":0,"source_error":0,"source_hit":2,"source_miss":0,"unsupported":0},"head":{"breaker_open":0,"filtered":0,"negative_cached":1,"source_error":0,"source_hit":0,"source_miss":0,"unsupported":0}},"pulled_bytes_total":4096,"pulled_objects_total":{"backfill":0,"background":0,"inline":1},"pull_failures_total":{"canceled":0,"etag_mismatch":0,"local_write":0,"queue_full":0,"source_access_denied":0,"source_connect":0,"source_not_found":0,"source_other":0,"source_server_error":0,"source_throttled":0,"source_timeout":1,"source_unsupported":0},"source_latency":{"buckets":[{"le_ms":5,"count":1},{"le_ms":10,"count":1},{"le_ms":20,"count":1},{"le_ms":50,"count":1},{"le_ms":100,"count":1},{"le_ms":200,"count":1},{"le_ms":500,"count":1},{"le_ms":1000,"count":2},{"le_ms":2000,"count":2},{"le_ms":5000,"count":2},{"le_ms":10000,"count":2},{"le_ms":20000,"count":2},{"le_ms":30000,"count":2},{"le_ms":60000,"count":2}],"count":3,"sum_ms":90753}},"last_source_error":{"class":"server_error","at":"2026-09-02T10:00:00Z"},"inflight_pulls":1,"queue_depth":1,"served_by_source_ratio":null,"updated_at":"2026-09-02T10:00:00Z","backfill":{"job_id":"11111111-1111-4111-8111-111111111111","state":"running","listed":2000,"enqueued":1500,"pulled":1400,"skipped_existing":500,"failed":3,"bytes":73400320,"updated_at":"2026-09-02T10:05:10Z"}}

crates/madmin/src/client.rs

@@ -400,7 +400,7 @@ impl AdminClient {
     }
 
     /// Signed POST returning a decoded JSON body.
-    async fn post_json<T: for<'de> Deserialize<'de>>(
+    pub(crate) async fn post_json<T: for<'de> Deserialize<'de>>(
         &self,
         path: &str,
         query: &[(&str, String)],

crates/madmin/src/on_demand_migration.rs

@@ -14,8 +14,9 @@
 
 //! On-Demand Migration admin API contract (ODM-07, rustfs/backlog#2154).
 //!
-//! Wire types for `PUT`/`GET`/`DELETE /v3/on-demand-migration/{bucket}` and
-//! `GET .../status`, mirroring the server's config model
+//! Wire types for `PUT`/`GET`/`DELETE /v3/on-demand-migration/{bucket}`,
+//! `GET .../status`, `POST .../backfill?op=start|cancel` and
+//! `GET .../backfill` (ODM-12), mirroring the server's config model
 //! (`crates/ecstore/src/bucket/on_demand_migration/config.rs`) and handler
 //! responses (`rustfs/src/admin/handlers/on_demand_migration.rs`). The SDK
 //! owns its own copies, madmin-go style; the fixtures under
@@ -34,6 +35,8 @@ pub const ON_DEMAND_MIGRATION_CONFIG_VERSION: u32 = 1;
 
 /// Query flag that validates and probes a config without saving it.
 const DRY_RUN_QUERY: &str = "dry-run";
+/// `POST .../backfill?op=` selector.
+const BACKFILL_OP_QUERY: &str = "op";
 
 /// Bucket-level on-demand migration configuration (request body of the
 /// `PUT`, redacted copy in every response).
@@ -300,7 +303,7 @@ pub struct OnDemandMigrationGetResponse {
 /// snapshot of the bucket. The runtime fields are `null` while the bucket has
 /// no live state on the answering node (module off, config absent or
 /// disabled); `provider` and `endpoint_host` then still describe the saved
-/// config, if any.
+/// config, if any. `backfill` is present once the bucket had a backfill job.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct OnDemandMigrationStatus {
     pub configured: bool,
@@ -328,6 +331,10 @@ pub struct OnDemandMigrationStatus {
     /// RFC 3339 save time of the config; `None` when not configured.
     #[serde(default)]
     pub updated_at: Option<String>,
+    /// Counters of the bucket's latest backfill job; absent until the bucket
+    /// has had one.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub backfill: Option<OnDemandMigrationBackfillSummary>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -385,6 +392,124 @@ pub struct OnDemandMigrationSourceError {
     pub at: String,
 }
 
+/// Lifecycle of a backfill job.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum OnDemandMigrationBackfillState {
+    Pending,
+    Running,
+    Paused,
+    Cancelled,
+    Completed,
+    CompletedWithFailures,
+    Failed,
+}
+
+impl OnDemandMigrationBackfillState {
+    /// Whether the job still runs (or is about to).
+    pub fn is_active(self) -> bool {
+        matches!(self, Self::Pending | Self::Running)
+    }
+}
+
+/// What to do with a listed key that already exists locally.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum OnDemandMigrationSkipExisting {
+    #[default]
+    Always,
+    EtagOrSize,
+}
+
+/// Body of `POST .../backfill?op=start`; every field is optional.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationBackfillRequest {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub prefix: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub skip_existing: Option<OnDemandMigrationSkipExisting>,
+    #[serde(default)]
+    pub dry_run: bool,
+}
+
+/// Last failure recorded by a backfill job; `key_hash` is a hash, never the key.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationBackfillError {
+    pub class: String,
+    #[serde(default)]
+    pub key_hash: Option<String>,
+    pub at: String,
+}
+
+/// Node running a backfill job and the lease it holds (RFC 3339).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationBackfillOwner {
+    pub node: String,
+    pub lease_until: String,
+}
+
+/// The backfill checkpoint as the server stores it. Field order is the
+/// on-disk and on-wire contract; unknown fields from newer servers are
+/// tolerated.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationBackfillJob {
+    pub format_version: u32,
+    pub job_id: String,
+    pub state: OnDemandMigrationBackfillState,
+    pub config_updated_at: String,
+    #[serde(default)]
+    pub prefix: Option<String>,
+    #[serde(default)]
+    pub skip_existing: OnDemandMigrationSkipExisting,
+    #[serde(default)]
+    pub dry_run: bool,
+    #[serde(default)]
+    pub continuation_token: Option<String>,
+    #[serde(default)]
+    pub listed: u64,
+    #[serde(default)]
+    pub enqueued: u64,
+    #[serde(default)]
+    pub pulled: u64,
+    #[serde(default)]
+    pub skipped_existing: u64,
+    #[serde(default)]
+    pub failed: u64,
+    #[serde(default)]
+    pub bytes: u64,
+    #[serde(default)]
+    pub last_key: Option<String>,
+    #[serde(default)]
+    pub last_error: Option<OnDemandMigrationBackfillError>,
+    #[serde(default)]
+    pub failed_keys: Vec<String>,
+    pub started_at: String,
+    pub updated_at: String,
+    #[serde(default)]
+    pub owner: Option<OnDemandMigrationBackfillOwner>,
+}
+
+/// `POST`/`GET .../backfill` response.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationBackfillResponse {
+    pub bucket: String,
+    pub job: OnDemandMigrationBackfillJob,
+}
+
+/// Counters of the latest backfill job, embedded in the status response.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationBackfillSummary {
+    pub job_id: String,
+    pub state: OnDemandMigrationBackfillState,
+    pub listed: u64,
+    pub enqueued: u64,
+    pub pulled: u64,
+    pub skipped_existing: u64,
+    pub failed: u64,
+    pub bytes: u64,
+    pub updated_at: String,
+}
+
 fn config_path(bucket: &str) -> String {
     format!("/v3/on-demand-migration/{}", percent_encode_path_segment(bucket))
 }
@@ -428,6 +553,43 @@ impl AdminClient {
     pub async fn on_demand_migration_status(&self, bucket: &str) -> Result<OnDemandMigrationStatus, AdminClientError> {
         self.get_json(&format!("{}/status", config_path(bucket))).await
     }
+
+    /// Start the backfill job of `bucket`. A job that still holds its lease
+    /// answers HTTP 409 `OnDemandMigrationBackfillRunning`.
+    pub async fn start_on_demand_migration_backfill(
+        &self,
+        bucket: &str,
+        request: &OnDemandMigrationBackfillRequest,
+    ) -> Result<OnDemandMigrationBackfillResponse, AdminClientError> {
+        let body = serde_json::to_vec(request).map_err(|err| AdminClientError::Decode {
+            message: err.to_string(),
+        })?;
+        self.post_json(&backfill_path(bucket), &[(BACKFILL_OP_QUERY, "start".to_string())], body)
+            .await
+    }
+
+    /// Cancel the backfill job of `bucket`; idempotent on a finished job. A
+    /// bucket that never had a job answers HTTP 404 `NoSuchBackfillJob`.
+    pub async fn cancel_on_demand_migration_backfill(
+        &self,
+        bucket: &str,
+    ) -> Result<OnDemandMigrationBackfillResponse, AdminClientError> {
+        self.post_json(&backfill_path(bucket), &[(BACKFILL_OP_QUERY, "cancel".to_string())], Vec::new())
+            .await
+    }
+
+    /// Read the backfill checkpoint of `bucket` (404 `NoSuchBackfillJob`
+    /// when none was ever started).
+    pub async fn on_demand_migration_backfill(
+        &self,
+        bucket: &str,
+    ) -> Result<OnDemandMigrationBackfillResponse, AdminClientError> {
+        self.get_json(&backfill_path(bucket)).await
+    }
+}
+
+fn backfill_path(bucket: &str) -> String {
+    format!("{}/backfill", config_path(bucket))
 }
 
 #[cfg(test)]
@@ -439,6 +601,8 @@ mod tests {
     const SET_RESPONSE_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/set_response.json");
     const GET_RESPONSE_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/get_response.json");
     const STATUS_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/status.json");
+    const STATUS_WITH_BACKFILL_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/status_with_backfill.json");
+    const BACKFILL_JOB_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/backfill_job.json");
 
     fn round_trip<T: Serialize + for<'de> Deserialize<'de>>(fixture: &str) -> T {
         let value: T = serde_json::from_str(fixture.trim()).expect("fixture decodes");
@@ -505,6 +669,7 @@ mod tests {
         assert_eq!(status.queue_depth, 1);
         assert_eq!(status.served_by_source_ratio, None, "the ratio is null, never a fabricated 0");
         assert_eq!(status.updated_at.as_deref(), Some("2026-09-02T10:00:00Z"));
+        assert!(status.backfill.is_none(), "a bucket without a job carries no summary");
     }
 
     #[test]
@@ -521,6 +686,108 @@ mod tests {
         assert_eq!(status.updated_at, None);
     }
 
+    #[test]
+    fn backfill_fixtures_round_trip_byte_for_byte() {
+        let response: OnDemandMigrationBackfillResponse = round_trip(BACKFILL_JOB_FIXTURE);
+        assert_eq!(response.bucket, "photos");
+        let job = &response.job;
+        assert_eq!(job.format_version, 1);
+        assert_eq!(job.state, OnDemandMigrationBackfillState::Running);
+        assert!(job.state.is_active());
+        assert_eq!(job.skip_existing, OnDemandMigrationSkipExisting::Always);
+        assert_eq!(job.continuation_token.as_deref(), Some("cGhvdG9zLzEwMDA="));
+        assert_eq!((job.listed, job.enqueued, job.pulled, job.failed), (2000, 1500, 1400, 3));
+        assert_eq!(job.failed_keys, vec!["9f2c3b0a1d4e5f60".to_string()]);
+        assert_eq!(job.last_error.as_ref().map(|e| e.class.as_str()), Some("source_timeout"));
+        assert_eq!(job.owner.as_ref().map(|o| o.node.as_str()), Some("node-a:9000"));
+
+        let status: OnDemandMigrationStatus = round_trip(STATUS_WITH_BACKFILL_FIXTURE);
+        let summary = status.backfill.expect("summary present");
+        assert_eq!(summary.job_id, job.job_id);
+        assert_eq!(summary.state, OnDemandMigrationBackfillState::Running);
+        assert_eq!(summary.bytes, 73_400_320);
+
+        // A newer server may add checkpoint fields; the client keeps decoding.
+        let newer =
+            BACKFILL_JOB_FIXTURE
+                .trim()
+                .replacen("\"listed\":2000", "\"listed\":2000,\"throttle_hint\":{\"mode\":\"soft\"}", 1);
+        let decoded: OnDemandMigrationBackfillResponse = serde_json::from_str(&newer).expect("unknown fields tolerated");
+        assert_eq!(decoded.job.listed, 2000);
+    }
+
+    #[test]
+    fn backfill_request_serializes_only_what_was_set() {
+        let minimal = serde_json::to_string(&OnDemandMigrationBackfillRequest::default()).expect("serialize");
+        assert_eq!(minimal, r#"{"dry_run":false}"#);
+        let full = serde_json::to_string(&OnDemandMigrationBackfillRequest {
+            prefix: Some("photos/".to_string()),
+            skip_existing: Some(OnDemandMigrationSkipExisting::EtagOrSize),
+            dry_run: true,
+        })
+        .expect("serialize");
+        assert_eq!(full, r#"{"prefix":"photos/","skip_existing":"etag_or_size","dry_run":true}"#);
+    }
+
+    #[tokio::test]
+    async fn backfill_start_cancel_and_get_use_the_registered_route() {
+        let server = TestServer::spawn(BACKFILL_JOB_FIXTURE, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+        let response = client
+            .start_on_demand_migration_backfill(
+                "photos",
+                &OnDemandMigrationBackfillRequest {
+                    prefix: Some("photos/".to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("start decodes");
+        assert_eq!(response.job.state, OnDemandMigrationBackfillState::Running);
+        let request = server.recorded();
+        assert_eq!(request.method, "POST");
+        assert_eq!(request.path, "/rustfs/admin/v3/on-demand-migration/photos/backfill");
+        assert_eq!(request.query, "op=start");
+        assert_eq!(request.header("content-type").as_deref(), Some("application/json"));
+        assert_eq!(request.body, r#"{"prefix":"photos/","dry_run":false}"#);
+
+        let server = TestServer::spawn(BACKFILL_JOB_FIXTURE, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+        client
+            .cancel_on_demand_migration_backfill("photos")
+            .await
+            .expect("cancel decodes");
+        let request = server.recorded();
+        assert_eq!(request.method, "POST");
+        assert_eq!(request.query, "op=cancel");
+        assert_eq!(request.body, "", "cancel sends no body");
+
+        let server = TestServer::spawn(BACKFILL_JOB_FIXTURE, 200).await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+        client.on_demand_migration_backfill("photos").await.expect("get decodes");
+        let request = server.recorded();
+        assert_eq!(request.method, "GET");
+        assert_eq!(request.path, "/rustfs/admin/v3/on-demand-migration/photos/backfill");
+
+        let server = TestServer::spawn(
+            r#"{"code":"OnDemandMigrationBackfillRunning","message":"a backfill job is already running"}"#,
+            409,
+        )
+        .await;
+        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
+        match client
+            .start_on_demand_migration_backfill("photos", &OnDemandMigrationBackfillRequest::default())
+            .await
+            .unwrap_err()
+        {
+            AdminClientError::HttpStatus { status, body } => {
+                assert_eq!(status, 409);
+                assert!(body.contains("OnDemandMigrationBackfillRunning"));
+            }
+            other => panic!("expected HttpStatus, got {other:?}"),
+        }
+    }
+
     #[test]
     fn minimal_config_expands_to_the_server_defaults() {
         let config = OnDemandMigrationConfig::new(OnDemandMigrationSource {

crates/obs/src/metrics/collectors/mod.rs

@@ -70,7 +70,8 @@ pub use notification::{NotificationStats, collect_notification_metrics};
 pub(crate) use notification_target::{NotificationTargetRuntimeStats, collect_notification_target_runtime_metrics};
 pub use notification_target::{NotificationTargetStats, collect_notification_target_metrics};
 pub use on_demand_migration::{
-    OnDemandMigrationBreakerState, OnDemandMigrationBucketStats, collect_on_demand_migration_metrics, source_latency_le_label,
+    OdmBackfillBucketStats, OdmBackfillRuntimeStats, OnDemandMigrationBreakerState, OnDemandMigrationBucketStats,
+    collect_on_demand_migration_backfill_metrics, collect_on_demand_migration_metrics, source_latency_le_label,
 };
 pub use replication::{ReplicationMetricsSnapshot, collect_replication_metrics};
 pub(crate) use replication::{ReplicationRuntimeStats, collect_replication_runtime_metrics};

crates/obs/src/metrics/collectors/on_demand_migration.rs

@@ -21,10 +21,12 @@
 
 use crate::metrics::report::PrometheusMetric;
 use crate::metrics::schema::on_demand_migration::{
-    BREAKER_STATE_CLOSED, BREAKER_STATE_HALF_OPEN, BREAKER_STATE_OPEN, BUCKET_L, LE_L, ODM_BREAKER_STATE_MD,
-    ODM_INFLIGHT_PULLS_MD, ODM_PULL_FAILURES_TOTAL_MD, ODM_PULLED_BYTES_TOTAL_MD, ODM_PULLED_OBJECTS_TOTAL_MD,
-    ODM_QUEUE_DEPTH_MD, ODM_REQUESTS_TOTAL_MD, ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD,
-    ODM_SOURCE_LATENCY_SECONDS_SUM_MD, OP_L, OUTCOME_L, PATH_L, REASON_L,
+    BREAKER_STATE_CLOSED, BREAKER_STATE_HALF_OPEN, BREAKER_STATE_OPEN, BUCKET_L, LE_L, ODM_BACKFILL_BYTES_MD,
+    ODM_BACKFILL_ENQUEUED_MD, ODM_BACKFILL_FAILED_MD, ODM_BACKFILL_JOBS_MD, ODM_BACKFILL_LISTED_MD, ODM_BACKFILL_PULLED_MD,
+    ODM_BACKFILL_SKIPPED_EXISTING_MD, ODM_BREAKER_STATE_MD, ODM_INFLIGHT_PULLS_MD, ODM_PULL_FAILURES_TOTAL_MD,
+    ODM_PULLED_BYTES_TOTAL_MD, ODM_PULLED_OBJECTS_TOTAL_MD, ODM_QUEUE_DEPTH_MD, ODM_REQUESTS_TOTAL_MD,
+    ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD, ODM_SOURCE_LATENCY_SECONDS_SUM_MD, OP_L,
+    OUTCOME_L, PATH_L, REASON_L, SERVER_L, STATE_L,
 };
 use std::borrow::Cow;
 use std::collections::BTreeMap;
@@ -170,6 +172,48 @@ pub fn collect_on_demand_migration_metrics(stats: &[OnDemandMigrationBucketStats
     metrics
 }
 
+/// Counters of one bucket's latest backfill job (ODM-12,
+/// rustfs/backlog#2159).
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct OdmBackfillBucketStats {
+    pub bucket: String,
+    /// Checkpoint state label (`running`, `completed`, ...).
+    pub state: String,
+    pub listed: u64,
+    pub enqueued: u64,
+    pub pulled: u64,
+    pub skipped_existing: u64,
+    pub failed: u64,
+    pub bytes: u64,
+}
+
+/// Backfill stats of every bucket with a checkpoint, labelled by node.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct OdmBackfillRuntimeStats {
+    pub server: String,
+    pub buckets: Vec<OdmBackfillBucketStats>,
+}
+
+/// Seven series per bucket: the state gauge and six counters.
+pub fn collect_on_demand_migration_backfill_metrics(stats: &OdmBackfillRuntimeStats) -> Vec<PrometheusMetric> {
+    let mut metrics = Vec::with_capacity(stats.buckets.len() * 7);
+    for bucket in &stats.buckets {
+        let labelled = |descriptor: &'static std::sync::LazyLock<crate::MetricDescriptor>, value: u64| {
+            PrometheusMetric::from_descriptor(descriptor, value as f64)
+                .with_label_owned(SERVER_L, stats.server.clone())
+                .with_label_owned(BUCKET_L, bucket.bucket.clone())
+        };
+        metrics.push(labelled(&ODM_BACKFILL_JOBS_MD, 1).with_label_owned(STATE_L, bucket.state.clone()));
+        metrics.push(labelled(&ODM_BACKFILL_LISTED_MD, bucket.listed));
+        metrics.push(labelled(&ODM_BACKFILL_ENQUEUED_MD, bucket.enqueued));
+        metrics.push(labelled(&ODM_BACKFILL_PULLED_MD, bucket.pulled));
+        metrics.push(labelled(&ODM_BACKFILL_SKIPPED_EXISTING_MD, bucket.skipped_existing));
+        metrics.push(labelled(&ODM_BACKFILL_FAILED_MD, bucket.failed));
+        metrics.push(labelled(&ODM_BACKFILL_BYTES_MD, bucket.bytes));
+    }
+    metrics
+}
+
 #[cfg(test)]
 pub(crate) mod tests {
     use super::*;
@@ -354,4 +398,81 @@ pub(crate) mod tests {
     fn empty_snapshot_emits_nothing() {
         assert!(collect_on_demand_migration_metrics(&[]).is_empty());
     }
+
+    /// One running and one completed job, as the scheduler would see them
+    /// from the local backfill runner.
+    pub(crate) fn backfill_golden_stats(server: &str) -> OdmBackfillRuntimeStats {
+        OdmBackfillRuntimeStats {
+            server: server.to_string(),
+            buckets: vec![
+                OdmBackfillBucketStats {
+                    bucket: "photos".to_string(),
+                    state: "running".to_string(),
+                    listed: 2000,
+                    enqueued: 1500,
+                    pulled: 1400,
+                    skipped_existing: 500,
+                    failed: 3,
+                    bytes: 73_400_320,
+                },
+                OdmBackfillBucketStats {
+                    bucket: "docs".to_string(),
+                    state: "completed".to_string(),
+                    ..Default::default()
+                },
+            ],
+        }
+    }
+
+    fn backfill_series<'a>(metrics: &'a [PrometheusMetric], name: &str, bucket: &str) -> Option<&'a PrometheusMetric> {
+        metrics.iter().find(|metric| {
+            metric.name == name
+                && metric
+                    .labels
+                    .iter()
+                    .any(|(label, value)| *label == BUCKET_L && value.as_ref() == bucket)
+        })
+    }
+
+    #[test]
+    fn backfill_collects_seven_series_per_bucket_with_the_odm_subsystem_prefix() {
+        let stats = backfill_golden_stats("node1:9000");
+        let metrics = collect_on_demand_migration_backfill_metrics(&stats);
+        assert_eq!(metrics.len(), 14);
+        assert!(
+            metrics
+                .iter()
+                .all(|metric| metric.name.starts_with("rustfs_on_demand_migration_backfill_"))
+        );
+        assert!(metrics.iter().all(|metric| {
+            metric
+                .labels
+                .iter()
+                .any(|(label, value)| *label == SERVER_L && value.as_ref() == "node1:9000")
+        }));
+
+        let jobs = backfill_series(&metrics, &ODM_BACKFILL_JOBS_MD.get_full_metric_name(), "photos").expect("jobs gauge");
+        assert_eq!(jobs.value, 1.0);
+        assert!(
+            jobs.labels
+                .iter()
+                .any(|(label, value)| *label == STATE_L && value.as_ref() == "running")
+        );
+        let listed = backfill_series(&metrics, &ODM_BACKFILL_LISTED_MD.get_full_metric_name(), "photos").expect("listed");
+        assert_eq!(listed.value, 2000.0);
+        let bytes = backfill_series(&metrics, &ODM_BACKFILL_BYTES_MD.get_full_metric_name(), "photos").expect("bytes");
+        assert_eq!(bytes.value, 73_400_320.0);
+        let docs_failed = backfill_series(&metrics, &ODM_BACKFILL_FAILED_MD.get_full_metric_name(), "docs").expect("docs failed");
+        assert_eq!(docs_failed.value, 0.0);
+        assert_eq!(
+            ODM_BACKFILL_LISTED_MD.get_full_metric_name(),
+            "rustfs_on_demand_migration_backfill_listed_total"
+        );
+    }
+
+    #[test]
+    fn backfill_no_buckets_means_no_series() {
+        let metrics = collect_on_demand_migration_backfill_metrics(&OdmBackfillRuntimeStats::default());
+        assert!(metrics.is_empty());
+    }
 }

crates/obs/src/metrics/mod.rs

@@ -34,6 +34,7 @@ pub(crate) use storage_api::metrics::{
     BucketOperations, BucketOptions, ObsBucketBandwidthMonitor, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore,
     StorageAdminApi, obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, obs_get_global_bucket_monitor,
     obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled,
-    obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_on_demand_migration_snapshot,
-    obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, obs_transition_state_handle,
+    obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_on_demand_migration_backfill_snapshot,
+    obs_on_demand_migration_snapshot, obs_replication_site_stats_snapshot, obs_resolve_object_store_handle,
+    obs_transition_state_handle,
 };

crates/obs/src/metrics/scheduler.rs

@@ -33,6 +33,7 @@ use crate::metrics::collectors::{
     NotificationStats,
     NotificationTargetRuntimeStats,
     NotificationTargetStats,
+    OdmBackfillRuntimeStats,
     OnDemandMigrationBucketStats,
     // System monitoring collectors (migrated from rustfs-obs::system)
     ProcessAttributeError,
@@ -65,6 +66,7 @@ use crate::metrics::collectors::{
     collect_node_metrics,
     collect_notification_runtime_metrics,
     collect_notification_target_runtime_metrics,
+    collect_on_demand_migration_backfill_metrics,
     collect_on_demand_migration_metrics,
     collect_process_attributes,
     collect_process_cpu_metrics,
@@ -121,12 +123,15 @@ use crate::metrics::schema::notification_target::{
     TARGET_ID as NOTIFICATION_TARGET_ID_LABEL, TARGET_TYPE as NOTIFICATION_TARGET_TYPE_LABEL,
 };
 use crate::metrics::schema::on_demand_migration::{
-    BUCKET_L as ODM_BUCKET_L, LE_L as ODM_LE_L, ODM_BREAKER_STATE_MD, ODM_INFLIGHT_PULLS_MD, ODM_PULL_FAILURES_TOTAL_MD,
+    BACKFILL_STATES as ODM_BACKFILL_STATES, BUCKET_L as ODM_BUCKET_L, LE_L as ODM_LE_L, ODM_BACKFILL_BYTES_MD,
+    ODM_BACKFILL_ENQUEUED_MD, ODM_BACKFILL_FAILED_MD, ODM_BACKFILL_JOBS_MD, ODM_BACKFILL_LISTED_MD, ODM_BACKFILL_PULLED_MD,
+    ODM_BACKFILL_SKIPPED_EXISTING_MD, ODM_BREAKER_STATE_MD, ODM_INFLIGHT_PULLS_MD, ODM_PULL_FAILURES_TOTAL_MD,
     ODM_PULLED_BYTES_TOTAL_MD, ODM_PULLED_OBJECTS_TOTAL_MD, ODM_QUEUE_DEPTH_MD, ODM_REQUESTS_TOTAL_MD,
     ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD, ODM_SOURCE_LATENCY_SECONDS_SUM_MD,
     OP_L as ODM_OP_L, OUTCOME_L as ODM_OUTCOME_L, PATH_L as ODM_PATH_L, PULL_FAILURE_REASONS as ODM_PULL_FAILURE_REASONS,
     PULL_PATHS as ODM_PULL_PATHS, REASON_L as ODM_REASON_L, REQUEST_OPS as ODM_REQUEST_OPS,
-    REQUEST_OUTCOMES as ODM_REQUEST_OUTCOMES, SOURCE_LATENCY_LE as ODM_SOURCE_LATENCY_LE,
+    REQUEST_OUTCOMES as ODM_REQUEST_OUTCOMES, SERVER_L as ODM_SERVER_L, SOURCE_LATENCY_LE as ODM_SOURCE_LATENCY_LE,
+    STATE_L as ODM_STATE_L,
 };
 use crate::metrics::schema::scanner::{
     BUCKET_LABEL as SCANNER_BUCKET_LABEL, CYCLE_SCOPE_LABEL as SCANNER_CYCLE_SCOPE_LABEL, DRIVE_LABEL as SCANNER_DRIVE_LABEL,
@@ -144,9 +149,9 @@ use crate::metrics::stats_collector::{
     collect_bucket_replication_stats_bundle, collect_bucket_stats, collect_cluster_and_health_stats,
     collect_cluster_config_stats, collect_cluster_usage_metric_stats, collect_compression_cluster_stats,
     collect_disk_and_system_drive_runtime_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats,
-    collect_ilm_runtime_metric_stats, collect_internode_network_stats, collect_on_demand_migration_stats,
-    collect_process_metric_bundle_with, collect_replication_stats, collect_scanner_runtime_metric_stats,
-    collect_system_cpu_and_memory_stats_with,
+    collect_ilm_runtime_metric_stats, collect_internode_network_stats, collect_on_demand_migration_backfill_stats,
+    collect_on_demand_migration_stats, collect_process_metric_bundle_with, collect_replication_stats,
+    collect_scanner_runtime_metric_stats, collect_system_cpu_and_memory_stats_with,
 };
 use crate::node_identity::{SERVER_LABEL, current_local_node_identity};
 use crate::telemetry::retire_metric_series;
@@ -914,6 +919,10 @@ fn on_demand_migration_bucket_live_keys(stats: &[OnDemandMigrationBucketStats])
     stats.iter().map(|stat| stat.bucket.clone()).collect()
 }
 
+fn on_demand_migration_backfill_bucket_live_keys(stats: &OdmBackfillRuntimeStats) -> HashSet<BucketKey> {
+    stats.buckets.iter().map(|stat| stat.bucket.clone()).collect()
+}
+
 fn update_series_zero_tombstones<T: Clone + Eq + std::hash::Hash>(
     has_seen_valid_snapshot: &mut bool,
     prev_live_keys: &mut HashSet<T>,
@@ -1646,6 +1655,42 @@ fn retire_on_demand_migration_metric_series(bucket: &str) -> usize {
         .sum()
 }
 
+/// Every on-demand migration backfill series one node's job can own. The
+/// `state` gauge is enumerated over the checkpoint's fixed lifecycle values
+/// because only one of them is emitted per cycle.
+fn on_demand_migration_backfill_metric_series(server: &str, bucket: &str) -> Vec<MetricSeriesKey> {
+    let job_labels = || {
+        vec![
+            (ODM_SERVER_L, Cow::Owned(server.to_string())),
+            (ODM_BUCKET_L, Cow::Owned(bucket.to_string())),
+        ]
+    };
+    let mut series = Vec::new();
+    for state in ODM_BACKFILL_STATES {
+        let mut labels = job_labels();
+        labels.push((ODM_STATE_L, Cow::Borrowed(state)));
+        series.push((ODM_BACKFILL_JOBS_MD.get_full_metric_name(), labels));
+    }
+    for descriptor in [
+        &*ODM_BACKFILL_LISTED_MD,
+        &*ODM_BACKFILL_ENQUEUED_MD,
+        &*ODM_BACKFILL_PULLED_MD,
+        &*ODM_BACKFILL_SKIPPED_EXISTING_MD,
+        &*ODM_BACKFILL_FAILED_MD,
+        &*ODM_BACKFILL_BYTES_MD,
+    ] {
+        series.push((descriptor.get_full_metric_name(), job_labels()));
+    }
+    series
+}
+
+fn retire_on_demand_migration_backfill_metric_series(server: &str, bucket: &str) -> usize {
+    on_demand_migration_backfill_metric_series(server, bucket)
+        .iter()
+        .map(|(name, labels)| retire_metric_series(name, labels))
+        .sum()
+}
+
 fn retire_repl_backlog_target_metric_series(bucket: &str, target_arn: &str) -> usize {
     let labels = [
         (BUCKET_L, Cow::Owned(bucket.to_string())),
@@ -2037,6 +2082,8 @@ pub fn init_metrics_runtime(token: CancellationToken) {
         let mut has_seen_proxy_bucket_snapshot = false;
         let mut prev_on_demand_migration_live_keys: HashSet<BucketKey> = HashSet::new();
         let mut has_seen_on_demand_migration_snapshot = false;
+        let mut prev_on_demand_migration_backfill_live_keys: HashSet<BucketKey> = HashSet::new();
+        let mut has_seen_on_demand_migration_backfill_snapshot = false;
         loop {
             tokio::select! {
                 _ = interval.tick() => {
@@ -2125,6 +2172,24 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                             prev_on_demand_migration_live_keys = current_on_demand_migration_live_keys;
                             has_seen_on_demand_migration_snapshot = true;
                             metrics.extend(collect_on_demand_migration_metrics(&on_demand_migration));
+                            // Backfill jobs come and go independently of the bucket's config,
+                            // so their series are retired on their own key set.
+                            let on_demand_migration_backfill = collect_on_demand_migration_backfill_stats();
+                            let current_on_demand_migration_backfill_live_keys =
+                                on_demand_migration_backfill_bucket_live_keys(&on_demand_migration_backfill);
+                            let retire_on_demand_migration_backfill_buckets = if has_seen_on_demand_migration_backfill_snapshot
+                            {
+                                prev_on_demand_migration_backfill_live_keys
+                                    .difference(&current_on_demand_migration_backfill_live_keys)
+                                    .cloned()
+                                    .collect::<Vec<_>>()
+                            } else {
+                                Vec::new()
+                            };
+                            prev_on_demand_migration_backfill_live_keys = current_on_demand_migration_backfill_live_keys;
+                            has_seen_on_demand_migration_backfill_snapshot = true;
+                            let on_demand_migration_backfill_server = on_demand_migration_backfill.server.clone();
+                            metrics.extend(collect_on_demand_migration_backfill_metrics(&on_demand_migration_backfill));
                             let replication = collect_replication_stats().await;
                             metrics.extend(collect_replication_runtime_metrics(&ReplicationRuntimeStats {
                                 server: current_local_node_identity(),
@@ -2154,6 +2219,12 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                             for bucket in retire_on_demand_migration_buckets {
                                 let _ = retire_on_demand_migration_metric_series(&bucket);
                             }
+                            for bucket in retire_on_demand_migration_backfill_buckets {
+                                let _ = retire_on_demand_migration_backfill_metric_series(
+                                    &on_demand_migration_backfill_server,
+                                    &bucket,
+                                );
+                            }
                         },
                     ).await;
                 }
@@ -2980,6 +3051,66 @@ mod tests {
         assert_eq!(on_demand_migration_metric_series("photos").len(), emitted.len());
     }
 
+    #[test]
+    fn on_demand_migration_backfill_bucket_keys_detect_finished_jobs() {
+        let stats = crate::metrics::collectors::on_demand_migration::tests::backfill_golden_stats("node1:9000");
+        let previous = on_demand_migration_backfill_bucket_live_keys(&stats);
+        let current = on_demand_migration_backfill_bucket_live_keys(&OdmBackfillRuntimeStats {
+            server: stats.server.clone(),
+            buckets: stats.buckets[..1].to_vec(),
+        });
+        let retired = previous.difference(&current).cloned().collect::<HashSet<_>>();
+
+        assert_eq!(retired, bucket_keys(&["docs"]));
+        assert_eq!(current, bucket_keys(&["photos"]));
+        assert!(on_demand_migration_backfill_bucket_live_keys(&OdmBackfillRuntimeStats::default()).is_empty());
+    }
+
+    /// Every emitted backfill series must be named by the retirement walk.
+    /// The walk is wider than one cycle's emission on purpose: only the
+    /// job's current `state` gauge is emitted, so retirement enumerates all
+    /// lifecycle values to clear whichever one is live.
+    #[test]
+    fn on_demand_migration_backfill_retirement_covers_every_emitted_series() {
+        let stats = crate::metrics::collectors::on_demand_migration::tests::backfill_golden_stats("node1:9000");
+        let emitted = collect_on_demand_migration_backfill_metrics(&stats)
+            .into_iter()
+            .map(|metric| {
+                (
+                    metric.name.to_string(),
+                    metric
+                        .labels
+                        .into_iter()
+                        .map(|(key, value)| (key, value.to_string()))
+                        .collect::<Vec<_>>(),
+                )
+            })
+            .collect::<HashSet<_>>();
+        let retired = stats
+            .buckets
+            .iter()
+            .flat_map(|bucket| on_demand_migration_backfill_metric_series(&stats.server, &bucket.bucket))
+            .map(|(name, labels)| {
+                (
+                    name,
+                    labels
+                        .into_iter()
+                        .map(|(key, value)| (key, value.to_string()))
+                        .collect::<Vec<_>>(),
+                )
+            })
+            .collect::<HashSet<_>>();
+
+        assert!(emitted.is_subset(&retired), "emitted: {emitted:?}, retired: {retired:?}");
+        assert_eq!(
+            on_demand_migration_backfill_metric_series("node1:9000", "photos").len(),
+            ODM_BACKFILL_STATES.len() + 6
+        );
+        // Without a process-global recorder there is nothing to retire; the
+        // walk itself must still cover every series.
+        assert_eq!(retire_on_demand_migration_backfill_metric_series("node1:9000", "photos"), 0);
+    }
+
     #[test]
     fn metrics_runtime_status_reports_disabled_state() {
         let snapshot = build_metrics_runtime_status_snapshot(false, false, fixed_metrics_runtime_config(), false);

crates/obs/src/metrics/schema/on_demand_migration.rs

@@ -35,6 +35,10 @@ pub const PATH_L: &str = "path";
 pub const REASON_L: &str = "reason";
 /// Upper bound (seconds) of a source latency bucket.
 pub const LE_L: &str = "le";
+/// Node the backfill job runs on.
+pub const SERVER_L: &str = "server";
+/// Lifecycle state of a backfill job.
+pub const STATE_L: &str = "state";
 
 /// Fixed `op` label values.
 pub const REQUEST_OPS: [&str; 2] = ["get", "head"];
@@ -71,6 +75,17 @@ pub const PULL_FAILURE_REASONS: [&str; 12] = [
 pub const SOURCE_LATENCY_LE: [&str; 15] = [
     "0.005", "0.01", "0.02", "0.05", "0.1", "0.2", "0.5", "1", "2", "5", "10", "20", "30", "60", "+Inf",
 ];
+/// Fixed `state` label values of `backfill_jobs`; mirrors the checkpoint's
+/// `BackfillState` variants in ecstore.
+pub const BACKFILL_STATES: [&str; 7] = [
+    "pending",
+    "running",
+    "paused",
+    "cancelled",
+    "completed",
+    "completed_with_failures",
+    "failed",
+];
 
 /// `breaker_state` gauge value: the breaker admits every request.
 pub const BREAKER_STATE_CLOSED: f64 = 0.0;
@@ -183,8 +198,80 @@ pub static ODM_BREAKER_STATE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     )
 });
 
-// backfill_* descriptors (`listed_total`, `pulled_total`, `skipped_total`,
-// `failed_total`, `state`) are added here by ODM-12 (rustfs/backlog#2159).
+// backfill_* descriptors (ODM-12, rustfs/backlog#2159). Unlike the request
+// path above, a backfill job runs on exactly one node at a time, so every
+// backfill series carries the owning node in `server` on top of `bucket`.
+
+const BACKFILL_JOBS: &str = "backfill_jobs";
+const BACKFILL_LISTED_TOTAL: &str = "backfill_listed_total";
+const BACKFILL_ENQUEUED_TOTAL: &str = "backfill_enqueued_total";
+const BACKFILL_PULLED_TOTAL: &str = "backfill_pulled_total";
+const BACKFILL_SKIPPED_EXISTING_TOTAL: &str = "backfill_skipped_existing_total";
+const BACKFILL_FAILED_TOTAL: &str = "backfill_failed_total";
+const BACKFILL_BYTES_TOTAL: &str = "backfill_bytes_total";
+
+pub static ODM_BACKFILL_JOBS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::from(BACKFILL_JOBS),
+        "On-demand migration backfill jobs by server, bucket and state (1 for the bucket's current state)",
+        &[SERVER_L, BUCKET_L, STATE_L],
+        subsystems::ON_DEMAND_MIGRATION,
+    )
+});
+
+pub static ODM_BACKFILL_LISTED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(BACKFILL_LISTED_TOTAL),
+        "Source keys listed by the on-demand migration backfill job, by server and bucket",
+        &[SERVER_L, BUCKET_L],
+        subsystems::ON_DEMAND_MIGRATION,
+    )
+});
+
+pub static ODM_BACKFILL_ENQUEUED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(BACKFILL_ENQUEUED_TOTAL),
+        "Keys queued for pulling by the on-demand migration backfill job, by server and bucket",
+        &[SERVER_L, BUCKET_L],
+        subsystems::ON_DEMAND_MIGRATION,
+    )
+});
+
+pub static ODM_BACKFILL_PULLED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(BACKFILL_PULLED_TOTAL),
+        "Objects stored locally by the on-demand migration backfill job, by server and bucket",
+        &[SERVER_L, BUCKET_L],
+        subsystems::ON_DEMAND_MIGRATION,
+    )
+});
+
+pub static ODM_BACKFILL_SKIPPED_EXISTING_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(BACKFILL_SKIPPED_EXISTING_TOTAL),
+        "Keys the on-demand migration backfill job skipped because a local object already existed, by server and bucket",
+        &[SERVER_L, BUCKET_L],
+        subsystems::ON_DEMAND_MIGRATION,
+    )
+});
+
+pub static ODM_BACKFILL_FAILED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(BACKFILL_FAILED_TOTAL),
+        "Keys the on-demand migration backfill job could not pull, by server and bucket",
+        &[SERVER_L, BUCKET_L],
+        subsystems::ON_DEMAND_MIGRATION,
+    )
+});
+
+pub static ODM_BACKFILL_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(BACKFILL_BYTES_TOTAL),
+        "Bytes stored locally by the on-demand migration backfill job, by server and bucket",
+        &[SERVER_L, BUCKET_L],
+        subsystems::ON_DEMAND_MIGRATION,
+    )
+});
 
 #[cfg(test)]
 mod tests {
@@ -208,6 +295,13 @@ mod tests {
             (&*ODM_SOURCE_LATENCY_SECONDS_SUM_MD, "source_latency_seconds_sum"),
             (&*ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, "source_latency_seconds_count"),
             (&*ODM_BREAKER_STATE_MD, "breaker_state"),
+            (&*ODM_BACKFILL_JOBS_MD, "backfill_jobs"),
+            (&*ODM_BACKFILL_LISTED_MD, "backfill_listed_total"),
+            (&*ODM_BACKFILL_ENQUEUED_MD, "backfill_enqueued_total"),
+            (&*ODM_BACKFILL_PULLED_MD, "backfill_pulled_total"),
+            (&*ODM_BACKFILL_SKIPPED_EXISTING_MD, "backfill_skipped_existing_total"),
+            (&*ODM_BACKFILL_FAILED_MD, "backfill_failed_total"),
+            (&*ODM_BACKFILL_BYTES_MD, "backfill_bytes_total"),
         ] {
             assert_eq!(descriptor.get_full_metric_name(), format!("rustfs_on_demand_migration_{suffix}"));
             assert_eq!(descriptor.subsystem, subsystems::ON_DEMAND_MIGRATION);
@@ -259,6 +353,23 @@ mod tests {
         assert_eq!(SOURCE_LATENCY_LE[14], "+Inf");
     }
 
+    #[test]
+    fn backfill_series_are_server_and_bucket_scoped() {
+        assert_eq!(ODM_BACKFILL_JOBS_MD.metric_type, MetricType::Gauge);
+        assert_eq!(labels(&ODM_BACKFILL_JOBS_MD), vec!["server", "bucket", "state"]);
+        for descriptor in [
+            &*ODM_BACKFILL_LISTED_MD,
+            &*ODM_BACKFILL_ENQUEUED_MD,
+            &*ODM_BACKFILL_PULLED_MD,
+            &*ODM_BACKFILL_SKIPPED_EXISTING_MD,
+            &*ODM_BACKFILL_FAILED_MD,
+            &*ODM_BACKFILL_BYTES_MD,
+        ] {
+            assert_eq!(descriptor.metric_type, MetricType::Counter);
+            assert_eq!(labels(descriptor), vec!["server", "bucket"]);
+        }
+    }
+
     #[test]
     fn fixed_label_values_are_unique() {
         for values in [
@@ -267,6 +378,7 @@ mod tests {
             PULL_PATHS.as_slice(),
             PULL_FAILURE_REASONS.as_slice(),
             SOURCE_LATENCY_LE.as_slice(),
+            BACKFILL_STATES.as_slice(),
         ] {
             let unique: std::collections::BTreeSet<_> = values.iter().collect();
             assert_eq!(unique.len(), values.len(), "{values:?}");

crates/obs/src/metrics/stats_collector.rs

@@ -26,15 +26,16 @@ use crate::metrics::collectors::{
     ClusterHealthStats, ClusterStats, ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats,
     DriveDetailedStats, DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats,
     IlmBackpressureStats, IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats,
-    OnDemandMigrationBucketStats, ProcessStats, ProcessStatusType, ReplicationMetricsSnapshot, ResourceStats,
-    ScannerRuntimeStats, ScannerStats,
+    OdmBackfillRuntimeStats, OnDemandMigrationBucketStats, ProcessStats, ProcessStatusType, ReplicationMetricsSnapshot,
+    ResourceStats, ScannerRuntimeStats, ScannerStats,
 };
 use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot};
 use crate::metrics::{
     BucketOperations, BucketOptions, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore, StorageAdminApi,
     obs_bucket_replication_stats_snapshot, obs_get_quota_config, obs_get_total_usable_capacity,
     obs_get_total_usable_capacity_free, obs_load_compression_total_from_memory, obs_load_data_usage_from_backend,
-    obs_on_demand_migration_snapshot, obs_replication_site_stats_snapshot, obs_resolve_object_store_handle,
+    obs_on_demand_migration_backfill_snapshot, obs_on_demand_migration_snapshot, obs_replication_site_stats_snapshot,
+    obs_resolve_object_store_handle,
 };
 use crate::node_identity::current_local_node_identity;
 use jiff::Timestamp;
@@ -667,6 +668,11 @@ pub fn collect_on_demand_migration_stats() -> Vec<OnDemandMigrationBucketStats>
     obs_on_demand_migration_snapshot()
 }
 
+/// Collect this node's on-demand migration backfill job progress.
+pub fn collect_on_demand_migration_backfill_stats() -> OdmBackfillRuntimeStats {
+    obs_on_demand_migration_backfill_snapshot(current_local_node_identity())
+}
+
 /// Collect site-level replication stats from the global replication runtime.
 pub async fn collect_replication_stats() -> ReplicationMetricsSnapshot {
     obs_site_replication_stats().await

crates/obs/src/metrics/storage_api.rs

@@ -17,6 +17,9 @@ use std::time::Duration;
 
 pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor;
 pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config;
+use rustfs_ecstore::api::bucket::on_demand_migration::backfill::{
+    BackfillCheckpoint as SourceBackfillCheckpoint, global_backfill_runner as source_global_backfill_runner,
+};
 use rustfs_ecstore::api::bucket::on_demand_migration::{
     BreakerState as SourceOdmBreakerState, OdmBucketSnapshot as SourceOdmBucketSnapshot,
     OnDemandMigrationSys as SourceOnDemandMigrationSys,
@@ -41,7 +44,9 @@ pub(crate) use rustfs_ecstore::api::runtime::{
 pub(crate) use rustfs_ecstore::api::storage::ECStore as ObsStore;
 use rustfs_storage_api as storage_contracts;
 
-use crate::metrics::collectors::{OnDemandMigrationBreakerState, OnDemandMigrationBucketStats};
+use crate::metrics::collectors::{
+    OdmBackfillBucketStats, OdmBackfillRuntimeStats, OnDemandMigrationBreakerState, OnDemandMigrationBucketStats,
+};
 
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct ObsBucketReplicationTargetStatsSnapshot {
@@ -496,6 +501,38 @@ pub(crate) fn obs_on_demand_migration_snapshot() -> Vec<OnDemandMigrationBucketS
         .collect()
 }
 
+fn on_demand_migration_backfill_stats_from_checkpoint(
+    bucket: String,
+    checkpoint: SourceBackfillCheckpoint,
+) -> OdmBackfillBucketStats {
+    OdmBackfillBucketStats {
+        bucket,
+        state: checkpoint.state.as_str().to_string(),
+        listed: checkpoint.listed,
+        enqueued: checkpoint.enqueued,
+        pulled: checkpoint.pulled,
+        skipped_existing: checkpoint.skipped_existing,
+        failed: checkpoint.failed,
+        bytes: checkpoint.bytes,
+    }
+}
+
+/// Backfill jobs running on this node, sorted by bucket. Empty until the
+/// runner is installed, and empty again once a job finishes: the series are
+/// per-node job progress, not a cluster-wide history.
+pub(crate) fn obs_on_demand_migration_backfill_snapshot(server: String) -> OdmBackfillRuntimeStats {
+    let buckets = source_global_backfill_runner()
+        .map(|runner| {
+            runner
+                .local_job_snapshots()
+                .into_iter()
+                .map(|(bucket, checkpoint)| on_demand_migration_backfill_stats_from_checkpoint(bucket, checkpoint))
+                .collect()
+        })
+        .unwrap_or_default();
+    OdmBackfillRuntimeStats { server, buckets }
+}
+
 pub(crate) async fn obs_replication_site_stats_snapshot(current_data_transfer_rate: f64) -> ObsReplicationSiteStatsSnapshot {
     let Some(stats) = get_global_replication_stats() else {
         return ObsReplicationSiteStatsSnapshot::default();
@@ -808,7 +845,8 @@ pub(crate) mod metrics {
         ObsBucketBandwidthMonitor, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore,
         obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_get_quota_config,
         obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled,
-        obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_on_demand_migration_snapshot,
-        obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, obs_transition_state_handle,
+        obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_on_demand_migration_backfill_snapshot,
+        obs_on_demand_migration_snapshot, obs_replication_site_stats_snapshot, obs_resolve_object_store_handle,
+        obs_transition_state_handle,
     };
 }