gantry+racer: pull benchmark, gantry retry/resume, slab resize recovery

cmd/gantry/agent_racer.go

@@ -85,7 +85,7 @@ func serveRacerAgent(ctx context.Context, c *config.Config, logger *slog.Logger,
 	// Always install the handler, even without a client, so unavailable requests
 	// retain Racer correlation headers and rate-limited failure diagnostics.
 	mirrorSrv := mirror.New(c, nil, originClient,
-		mirror.WithContentBackend(racer.NewHandler(deps.client, originClient, logger)), mirror.WithLogger(logger), mirror.WithStartupReadinessGate())
+		mirror.WithContentBackend(racer.NewHandler(deps.client, originClient, logger, racer.WithRetryBudget(c.RacerRetryBudget), racer.WithRetryObserver(telemetry.MirrorRetry))), mirror.WithLogger(logger), mirror.WithStartupReadinessGate())
 
 	var (
 		ready   atomic.Bool

cmd/racer-dataplane/alloc/src/slab.rs

@@ -64,6 +64,9 @@ pub struct Slab<C: Charge> {
 
     max_record_bytes: usize,
 
+    /// Recreate a regular slab file whose size differs from the configured capacity.
+    reset_mismatched: bool,
+
     opened: RefCell<Option<OpenSlab>>,
 
     writes: Rc<WriteState>,
@@ -84,12 +87,23 @@ impl<C: Charge> Slab<C> {
             capacity_bytes,
             segment_bytes,
             max_record_bytes,
+            reset_mismatched: false,
             opened: RefCell::new(None),
             writes: Rc::new(WriteState::default()),
             idle_buffer: Rc::new(RefCell::new(crate::BufferPool::default())),
         }
     }
 
+    /// Treat an existing regular slab file of the wrong size as a disposable cache:
+    /// truncate it to zero and recreate it at the configured capacity instead of
+    /// failing with `InvalidConfiguration`. Device-backed slabs ignore this flag.
+    /// The caller must ensure no recovered metadata references the old contents.
+    #[must_use]
+    pub fn reset_mismatched_size(mut self) -> Self {
+        self.reset_mismatched = true;
+        self
+    }
+
     /// Describe device ranges in logical segment order, without creating, sizing,
     /// or locking files. Bounds use BLKGETSIZE64 for block devices and file length
     /// for regular direct files. The caller must supply suitable alignment and keep
@@ -417,8 +431,16 @@ impl<C: Charge> Slab<C> {
             // The virtual filesystem has a fixed root owner, independent of host uid.
             validate_file(stat.stx_mode as u32, stat.stx_uid, 0, stat.stx_nlink as u64)?;
             let a = alignment_from_stat(&stat)?;
-            let geometry = self.validate_layout(a, stat.stx_size)?;
-            if stat.stx_size == 0 {
+            let mut size = stat.stx_size;
+            if self.reset_mismatched && size != 0 && size != self.capacity_bytes {
+                self.validate_layout(a, 0)?;
+                handle
+                    .set_len(0)
+                    .map_err(|e| system_error("ftruncate", e))?;
+                size = 0;
+            }
+            let geometry = self.validate_layout(a, size)?;
+            if size == 0 {
                 handle
                     .set_len(self.capacity_bytes)
                     .map_err(|e| system_error("ftruncate", e))?;
@@ -449,7 +471,7 @@ impl<C: Charge> Slab<C> {
             unsafe { libc::geteuid() },
             metadata.nlink(),
         )?;
-        let size = metadata.len();
+        let mut size = metadata.len();
         // Delay O_DIRECT until after rejecting nonregular files (including FIFOs).
         // SAFETY: fcntl synchronously borrows this live descriptor.
         let flags = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_GETFL) };
@@ -470,6 +492,17 @@ impl<C: Charge> Slab<C> {
             ));
         }
         let a = probe(&file)?;
+        if self.reset_mismatched && size != 0 && size != self.capacity_bytes {
+            // Validate everything except the stale size before discarding data.
+            self.validate_layout(a, 0)?;
+            eprintln!(
+                "racer: slab {} has {size} bytes, configured {}; discarding stale cache contents",
+                path.display(),
+                self.capacity_bytes
+            );
+            file.set_len(0).map_err(|e| system_error("ftruncate", e))?;
+            size = 0;
+        }
         let geometry = self.validate_layout(a, size)?;
         if size == 0 {
             file.set_len(self.capacity_bytes)
@@ -1871,6 +1904,62 @@ mod tests {
         assert_eq!(std::fs::metadata(path).unwrap().len(), capacity);
     }
 
+    /// Opt-in reset recreates stale-size cache files but still rejects invalid dimensions.
+    #[test]
+    fn reset_mismatched_size_recreates_grown_or_shrunk_files() {
+        use std::os::unix::fs::PermissionsExt;
+        let directory = Directory::new();
+        let probe = Slab::<()>::new(directory.0.join("capability-probe"), 4096, 4096, 512);
+        if real_alignment(probe.open_now()).is_none() {
+            return;
+        }
+        for (old, new) in [(8192u64, 16384u64), (16384, 8192)] {
+            let path = directory.0.join(format!("data-{old}-{new}"));
+            std::fs::write(&path, vec![42u8; old as usize]).unwrap();
+            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();
+            // Without opt-in the stale size is still rejected and preserved.
+            let strict = Slab::<()>::new(path.clone(), new, 4096, 512);
+            assert_eq!(strict.open_now(), Err(Error::InvalidConfiguration));
+            assert_eq!(std::fs::metadata(&path).unwrap().len(), old);
+            drop(strict);
+            let slab = Slab::<()>::new(path.clone(), new, 4096, 512).reset_mismatched_size();
+            assert!(real_alignment(slab.open_now()).is_some());
+            assert_eq!(slab.capacity_bytes(), new);
+            drop(slab);
+            let contents = std::fs::read(&path).unwrap();
+            assert_eq!(contents.len() as u64, new);
+            // Stale cache bytes are discarded rather than reinterpreted.
+            assert!(contents.iter().all(|b| *b == 0));
+        }
+        // Matching sizes keep their contents.
+        let path = directory.0.join("same");
+        std::fs::write(&path, vec![7u8; 8192]).unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();
+        let slab = Slab::<()>::new(path.clone(), 8192, 4096, 512).reset_mismatched_size();
+        assert!(real_alignment(slab.open_now()).is_some());
+        drop(slab);
+        assert!(std::fs::read(&path).unwrap().iter().all(|b| *b == 7));
+        // Genuinely invalid configurations still fail and leave stale files untouched.
+        let path = directory.0.join("invalid-existing");
+        std::fs::write(&path, vec![9u8; 8192]).unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();
+        for (capacity, segment, record) in [
+            (0, 4096, 512),
+            (4096, 0, 512),
+            (4096, 4096, 0),
+            (12289, 4096, 512),
+            (16384, 4096, 4097),
+        ] {
+            assert_eq!(
+                Slab::<()>::new(path.clone(), capacity, segment, record)
+                    .reset_mismatched_size()
+                    .open_now(),
+                Err(Error::InvalidConfiguration)
+            );
+            assert_eq!(std::fs::read(&path).unwrap(), vec![9u8; 8192]);
+        }
+    }
+
     /// Invalid startup dimensions never truncate a preexisting nonempty file.
     #[test]
     fn open_rejects_bad_layout_and_existing_size_without_truncating() {

cmd/racer-dataplane/src/app.rs

@@ -243,6 +243,81 @@ impl NodeState {
     }
 }
 
+/// Remove slab files left by workers that no longer exist after a worker-count change.
+///
+/// Only regular files named `worker-<id>-slab-0.dat` with an id outside `workers` are
+/// removed. Each file is removed only while holding the same exclusive lock that slab
+/// startup takes, and only if the path still names the locked inode, so a concurrent
+/// instance's live slabs are skipped. Failures are logged and ignored because stale
+/// files never affect correctness.
+fn prune_stale_slab_files(directory: &std::path::Path, workers: &[WorkerId]) {
+    use std::os::fd::AsRawFd;
+    use std::os::unix::fs::MetadataExt;
+    use std::os::unix::fs::OpenOptionsExt;
+
+    let Ok(entries) = std::fs::read_dir(directory) else {
+        return;
+    };
+    for entry in entries.flatten() {
+        let name = entry.file_name();
+        let Some(id) = name
+            .to_str()
+            .and_then(|n| n.strip_prefix("worker-"))
+            .and_then(|n| n.strip_suffix("-slab-0.dat"))
+            .and_then(|n| n.parse::<u16>().ok())
+        else {
+            continue;
+        };
+        if workers.contains(&WorkerId(id)) || !entry.file_type().is_ok_and(|t| t.is_file()) {
+            continue;
+        }
+        let path = entry.path();
+        let file = match std::fs::OpenOptions::new()
+            .read(true)
+            .custom_flags(libc::O_NOFOLLOW)
+            .open(&path)
+        {
+            Ok(file) => file,
+            Err(error) => {
+                eprintln!(
+                    "racer-dataplane: stage=slabs failed to open stale slab file {}: {error}",
+                    path.display()
+                );
+                continue;
+            }
+        };
+        // SAFETY: flock synchronously borrows this live descriptor.
+        if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
+            eprintln!(
+                "racer-dataplane: stage=slabs kept locked stale slab file {}: {}",
+                path.display(),
+                std::io::Error::last_os_error()
+            );
+            continue;
+        }
+        let same = match (file.metadata(), std::fs::symlink_metadata(&path)) {
+            (Ok(locked), Ok(current)) => {
+                current.is_file() && locked.dev() == current.dev() && locked.ino() == current.ino()
+            }
+            _ => false,
+        };
+        if !same {
+            continue;
+        }
+        // The lock is held until `file` drops, after the unlink.
+        match std::fs::remove_file(&path) {
+            Ok(()) => eprintln!(
+                "racer-dataplane: stage=slabs removed stale slab file {}",
+                path.display()
+            ),
+            Err(error) => eprintln!(
+                "racer-dataplane: stage=slabs failed to remove stale slab file {}: {error}",
+                path.display()
+            ),
+        }
+    }
+}
+
 impl Application {
     /// Composition only. Does not open files, spawn threads, or accept requests.
     pub fn assemble(config: Config) -> Result<Self> {
@@ -346,6 +421,10 @@ impl Application {
                 .map(devices::Plan::claimed_disks)
                 .unwrap_or_default(),
         });
+        if devices.is_none() {
+            let workers: Vec<_> = plan.pairs.iter().map(|p| p.worker).collect();
+            prune_stale_slab_files(&self.config.slab_directory, &workers);
+        }
         node.devices = devices;
         Arc::get_mut(&mut self.config)
             .ok_or(Error::InvalidConfiguration)?
@@ -1491,6 +1570,13 @@ impl WorkerApplication {
                 config.segment_bytes,
                 crate::model::PAGE_BYTES as usize + crate::store::MAX_HEADER_BYTES + 16,
             )
+            // Slab files are a re-fetchable cache. A checkpoint taken at the old
+            // size no longer matches slab_bytes, so recovery starts this worker
+            // cold. An older checkpoint taken at the new size, as after a resize
+            // rollback, may still be accepted; record validation and AEAD then
+            // turn its references to discarded pages into misses, as for any
+            // checkpoint that outlives its payload.
+            .reset_mismatched_size()
         });
         let reader = Rc::new(
             StoreReader::new(

cmd/racer-dataplane/src/app/recovery.rs

@@ -1001,6 +1001,55 @@ mod tests {
         );
     }
 
+    /// A geometry change (slab size, segment size, or worker count) skips stale
+    /// checkpoints and starts cold instead of failing node recovery.
+    #[test]
+    fn changed_slab_size_or_worker_count_skips_stale_checkpoint() {
+        let node = NodeState::default();
+        let keys = crate::test_support::security::keys();
+        let live = |ids: &[WorkerId], g: CheckpointGeometry| -> HashMap<_, _> {
+            ids.iter().map(|id| (*id, g)).collect()
+        };
+        let choose = |cuts: Vec<CheckpointImage>, g: &HashMap<WorkerId, CheckpointGeometry>| {
+            select(cuts, g, &node.workers, (4, 4), &keys, &caches())
+        };
+        let three = [WorkerId(0), WorkerId(1), WorkerId(2)];
+        let five = [
+            WorkerId(0),
+            WorkerId(1),
+            WorkerId(2),
+            WorkerId(3),
+            WorkerId(4),
+        ];
+        // Sanity: an unchanged geometry and worker set recovers.
+        assert!(choose(vec![image(1, &three)], &live(&three, geometry())).is_some());
+        // Larger slab per worker.
+        let bigger =
+            CheckpointGeometry::new(16384, 4096, 4, Alignment::new(4096, 4096, 4096).unwrap())
+                .unwrap();
+        assert!(choose(vec![image(1, &three)], &live(&three, bigger)).is_none());
+        // Different segment size.
+        let segment =
+            CheckpointGeometry::new(8192, 8192, 1, Alignment::new(4096, 4096, 4096).unwrap())
+                .unwrap();
+        assert!(choose(vec![image(1, &three)], &live(&three, segment)).is_none());
+        // More workers, with and without a slab size change.
+        assert!(choose(vec![image(1, &three)], &live(&five, geometry())).is_none());
+        assert!(choose(vec![image(1, &three)], &live(&five, bigger)).is_none());
+        // Fewer workers.
+        assert!(choose(vec![image(1, &five)], &live(&three, geometry())).is_none());
+        // A newer compatible cut still wins over older stale ones.
+        assert_eq!(
+            choose(
+                vec![image(3, &five), image(2, &three)],
+                &live(&three, geometry())
+            )
+            .unwrap()
+            .sequence,
+            2
+        );
+    }
+
     #[test]
     fn newest_complete_cut_wins_and_partial_duplicate_or_foreign_workers_fall_back() {
         let node = NodeState::default();

cmd/racer-dataplane/src/app/tests.rs

@@ -4880,3 +4880,73 @@ fn snapshot_refresh_applies_priority_live_and_reweighs_memory() {
         );
     }
 }
+
+/// Slab files for workers outside the current plan are removed; others are kept.
+#[test]
+fn prune_stale_slab_files_removes_only_extra_worker_files() {
+    let directory = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("target")
+        .join(format!("prune-slabs-test-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&directory);
+    std::fs::create_dir_all(&directory).unwrap();
+    for name in [
+        "worker-0-slab-0.dat",
+        "worker-1-slab-0.dat",
+        "worker-2-slab-0.dat",
+        "worker-3-slab-0.dat",
+        "worker-4-slab-0.dat",
+        "checkpoint.0",
+        "worker-x-slab-0.dat",
+    ] {
+        std::fs::write(directory.join(name), b"x").unwrap();
+    }
+    std::fs::create_dir(directory.join("worker-9-slab-0.dat")).unwrap();
+    super::prune_stale_slab_files(&directory, &[WorkerId(0), WorkerId(1), WorkerId(2)]);
+    let mut left: Vec<_> = std::fs::read_dir(&directory)
+        .unwrap()
+        .map(|e| e.unwrap().file_name().into_string().unwrap())
+        .collect();
+    left.sort();
+    assert_eq!(
+        left,
+        [
+            "checkpoint.0",
+            "worker-0-slab-0.dat",
+            "worker-1-slab-0.dat",
+            "worker-2-slab-0.dat",
+            "worker-9-slab-0.dat",
+            "worker-x-slab-0.dat",
+        ]
+    );
+    // A missing directory is not an error.
+    super::prune_stale_slab_files(&directory.join("missing"), &[WorkerId(0)]);
+    std::fs::remove_dir_all(&directory).unwrap();
+}
+
+/// A stale slab file still locked by another instance is kept.
+#[test]
+fn prune_stale_slab_files_skips_locked_files() {
+    use std::os::fd::AsRawFd;
+    let directory = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("target")
+        .join(format!("prune-locked-slabs-test-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&directory);
+    std::fs::create_dir_all(&directory).unwrap();
+    let locked = directory.join("worker-3-slab-0.dat");
+    let unlocked = directory.join("worker-4-slab-0.dat");
+    std::fs::write(&locked, b"x").unwrap();
+    std::fs::write(&unlocked, b"x").unwrap();
+    let holder = std::fs::File::open(&locked).unwrap();
+    // SAFETY: flock borrows a live descriptor owned by this test.
+    assert_eq!(
+        unsafe { libc::flock(holder.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) },
+        0
+    );
+    super::prune_stale_slab_files(&directory, &[WorkerId(0)]);
+    assert!(locked.exists());
+    assert!(!unlocked.exists());
+    drop(holder);
+    super::prune_stale_slab_files(&directory, &[WorkerId(0)]);
+    assert!(!locked.exists());
+    std::fs::remove_dir_all(&directory).unwrap();
+}

cmd/racer-loadgen/gantry_test.go

@@ -103,7 +103,9 @@ func TestGantryIntegration(t *testing.T) {
 		return metadata, body, err
 	})
 
-	server := httptest.NewServer(mirror.New(cfg, nil, upstream, mirror.WithContentBackend(gantryracer.NewHandler(client, upstream, nil))).Handler())
+	// Retries are disabled so callback counts reflect exactly one attempt per page.
+	handler := gantryracer.NewHandler(client, upstream, nil, gantryracer.WithRetryBudget(0))
+	server := httptest.NewServer(mirror.New(cfg, nil, upstream, mirror.WithContentBackend(handler)).Handler())
 	t.Cleanup(server.Close)
 	opts := pullTestOptions(server.URL)
 	opts.Namespace = "loadgen.invalid"