Add kumo.job() for scheduled job scripts

3rdparty/notices/inventory.json

@@ -515,6 +515,7 @@
         "rust:cpp-demangle@0.4.5",
         "rust:cpufeatures@0.2.17",
         "rust:cpufeatures@0.3.0",
+        "rust:crc32fast@1.5.0",
         "rust:crossbeam-deque@0.8.6",
         "rust:crossbeam-epoch@0.9.21",
         "rust:crossbeam-utils@0.8.21",
@@ -3642,7 +3643,8 @@
     },
     {
       "artifact_ids": [
-        "diskgraph-extension"
+        "diskgraph-extension",
+        "diskgraph-ingest"
       ],
       "copyright": [
         "Copyright (c) 2018 Sam Rijs, Alex Crichton and contributors"

packages/diskgraph/diskgraph-ingest/Cargo.toml

@@ -55,6 +55,7 @@ epserde = { workspace = true }
 memmap2 = "0.9"
 libc = "0.2"
 rayon = "1"
+crc32fast = "1.5"
 jemallocator = { version = "0.5", optional = true, features = ["profiling"] }
 jemalloc-ctl = { version = "0.5", optional = true }
 sysinfo = "0.30"

packages/diskgraph/diskgraph-ingest/src/snapshot_mirror.rs

@@ -36,6 +36,7 @@ use anyhow::{Context, Result};
 use object_store::path::Path as ObjectPath;
 use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
 
 use diskgraph_fs::{CloudTransfer, CloudTransferTicket};
 
@@ -488,21 +489,9 @@ pub(crate) fn validate_manifest_rel_path(rel: &str) -> Result<(), &'static str>
     Ok(())
 }
 
-/// Verify the snapshot rooted at `output_dir` against its `_MANIFEST.json`.
-/// Used by the consumer side after a Python-driven dataplane→local copy
-/// completes, to catch lifecycle-mid-delete races and partial copies
-/// before the snapshot is handed to a workload.
-///
-/// Pass on success; [`CorruptionError`] on any deviation from the
-/// manifest contract. Errors are precise about what failed so the
-/// caller's alerting carries actionable detail (which file is missing,
-/// expected vs actual size, etc.).
-pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::Result<(), CorruptionError> {
-    let success_path = output_dir.join(SUCCESS_MARKER_FILENAME);
-    if !success_path.exists() {
-        return Err(CorruptionError::MissingSuccess(success_path));
-    }
-
+/// Read and validate `_MANIFEST.json` at `output_dir`: schema version and
+/// every rel_path's confinement to the snapshot root.
+fn read_manifest(output_dir: &std::path::Path) -> std::result::Result<Manifest, CorruptionError> {
     let manifest_path = output_dir.join(MANIFEST_FILENAME);
     let manifest_bytes = match std::fs::read(&manifest_path) {
         Ok(b) => b,
@@ -548,6 +537,25 @@ pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::R
             });
         }
     }
+    Ok(manifest)
+}
+
+/// Verify the snapshot rooted at `output_dir` against its `_MANIFEST.json`.
+/// Used by the consumer side after a Python-driven dataplane→local copy
+/// completes, to catch lifecycle-mid-delete races and partial copies
+/// before the snapshot is handed to a workload.
+///
+/// Pass on success; [`CorruptionError`] on any deviation from the
+/// manifest contract. Errors are precise about what failed so the
+/// caller's alerting carries actionable detail (which file is missing,
+/// expected vs actual size, etc.).
+pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::Result<(), CorruptionError> {
+    let success_path = output_dir.join(SUCCESS_MARKER_FILENAME);
+    if !success_path.exists() {
+        return Err(CorruptionError::MissingSuccess(success_path));
+    }
+
+    let manifest = read_manifest(output_dir)?;
 
     let mut missing: Vec<String> = Vec::new();
     let mut size_mismatches: Vec<(String, u64, u64)> = Vec::new();
@@ -584,6 +592,71 @@ pub fn verify_snapshot_integrity(output_dir: &std::path::Path) -> std::result::R
     Ok(())
 }
 
+/// CRC-32 of every `range_bytes` span of each file `_MANIFEST.json` lists,
+/// and of `_MANIFEST.json` itself, keyed by rel_path (an empty file has one
+/// zero entry). Lets a copy of the snapshot be checked for content as well
+/// as size; the manifest format is unchanged.
+pub fn snapshot_range_checksums(
+    output_dir: &std::path::Path,
+    range_bytes: u64,
+) -> std::result::Result<BTreeMap<String, Vec<u32>>, CorruptionError> {
+    use rayon::prelude::*;
+    use std::os::unix::fs::FileExt;
+
+    const CHUNK: u64 = 1 << 20;
+    if range_bytes == 0 {
+        return Err(CorruptionError::Io {
+            path: output_dir.to_path_buf(),
+            detail: "range_bytes must be positive".into(),
+        });
+    }
+    let mut files = read_manifest(output_dir)?.files;
+    let manifest_size = std::fs::metadata(output_dir.join(MANIFEST_FILENAME))
+        .map(|meta| meta.len())
+        .unwrap_or(0);
+    files.push(ManifestEntry {
+        rel_path: MANIFEST_FILENAME.into(),
+        size_bytes: manifest_size,
+    });
+    let ranges: Vec<(usize, u64)> = files
+        .iter()
+        .enumerate()
+        .flat_map(|(index, entry)| {
+            let count = entry.size_bytes.div_ceil(range_bytes).max(1);
+            (0..count).map(move |range| (index, range))
+        })
+        .collect();
+    let crcs = ranges
+        .par_iter()
+        .map(|&(index, range)| {
+            let entry = &files[index];
+            let path = output_dir.join(&entry.rel_path);
+            let io = |e: std::io::Error| CorruptionError::Io {
+                path: path.clone(),
+                detail: e.to_string(),
+            };
+            let file = std::fs::File::open(&path).map_err(io)?;
+            let start = range * range_bytes;
+            let end = entry.size_bytes.min(start + range_bytes);
+            let mut hasher = crc32fast::Hasher::new();
+            let mut buffer = vec![0u8; CHUNK.min(end - start) as usize];
+            let mut offset = start;
+            while offset < end {
+                let length = CHUNK.min(end - offset) as usize;
+                file.read_exact_at(&mut buffer[..length], offset).map_err(io)?;
+                hasher.update(&buffer[..length]);
+                offset += length as u64;
+            }
+            Ok(hasher.finalize())
+        })
+        .collect::<std::result::Result<Vec<u32>, CorruptionError>>()?;
+    let mut out: BTreeMap<String, Vec<u32>> = BTreeMap::new();
+    for (&(index, _), crc) in ranges.iter().zip(crcs) {
+        out.entry(files[index].rel_path.clone()).or_default().push(crc);
+    }
+    Ok(out)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -933,4 +1006,35 @@ mod tests {
         verify_snapshot_integrity(local_root.path())?;
         Ok(())
     }
+
+    #[test]
+    fn range_checksums_cover_every_listed_file_and_the_manifest() {
+        let tmp = tempfile::tempdir().unwrap();
+        let big: Vec<u8> = (0..10u32).flat_map(|i| i.to_le_bytes()).collect();
+        write_marker_files(tmp.path(), &[("a.bin", &big), ("nested/empty", b"")]);
+        write_manifest(
+            tmp.path(),
+            vec![
+                ManifestEntry {
+                    rel_path: "a.bin".into(),
+                    size_bytes: 40,
+                },
+                ManifestEntry {
+                    rel_path: "nested/empty".into(),
+                    size_bytes: 0,
+                },
+            ],
+        );
+        let crcs = snapshot_range_checksums(tmp.path(), 16).unwrap();
+        let manifest = std::fs::read(tmp.path().join(MANIFEST_FILENAME)).unwrap();
+        assert_eq!(crcs["a.bin"], [&big[..16], &big[16..32], &big[32..]].map(crc32fast::hash));
+        assert_eq!(crcs["nested/empty"], vec![0]);
+        assert_eq!(crcs[MANIFEST_FILENAME], manifest.chunks(16).map(crc32fast::hash).collect::<Vec<_>>());
+
+        std::fs::write(tmp.path().join("a.bin"), &big[..20]).unwrap();
+        assert!(matches!(
+            snapshot_range_checksums(tmp.path(), 16).unwrap_err(),
+            CorruptionError::Io { .. }
+        ));
+    }
 }

packages/diskgraph/diskgraph-pybind/build.rs

@@ -9,8 +9,37 @@ mod native_cpu_guard {
     ));
 }
 
+fn build_commit() -> String {
+    println!("cargo:rerun-if-env-changed=DISKGRAPH_BUILD_COMMIT");
+    if let Ok(commit) = std::env::var("DISKGRAPH_BUILD_COMMIT") {
+        return commit;
+    }
+    let git = |args: &[&str]| {
+        std::process::Command::new("git")
+            .args(args)
+            .output()
+            .ok()
+            .filter(|out| out.status.success())
+            .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_owned())
+            .filter(|out| !out.is_empty())
+    };
+    for name in ["HEAD", "packed-refs"] {
+        if let Some(path) = git(&["rev-parse", "--git-path", name]) {
+            println!("cargo:rerun-if-changed={path}");
+        }
+    }
+    if let Some(head) = git(&["symbolic-ref", "-q", "HEAD"]) {
+        if let Some(path) = git(&["rev-parse", "--git-path", &head]) {
+            println!("cargo:rerun-if-changed={path}");
+        }
+    }
+    git(&["rev-parse", "--short=12", "HEAD"]).unwrap_or_else(|| "unknown".into())
+}
+
 fn main() {
     pyo3_build_config::use_pyo3_cfgs();
+    println!("cargo:rerun-if-changed=build.rs");
+    println!("cargo:rustc-env=DISKGRAPH_BUILD_COMMIT={}", build_commit());
 
     if std::env::var_os("CARGO_FEATURE_EXTENSION_MODULE").is_some() {
         native_cpu_guard::reject_native_cpu_build("diskgraph Python extension");

packages/diskgraph/diskgraph-pybind/python/diskgraph/__init__.py

@@ -3,6 +3,8 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from ._c import (
+    __version__,
+    BUILD_COMMIT,
     export_tt_artifacts,
     convert_kumo_config,
     ingest,
@@ -38,6 +40,7 @@
     write_success_marker,
     touch_success_marker,
     verify_snapshot_integrity,
+    snapshot_range_checksums,
     has_resumable_ingest,
     validated_index_resume_progress,
     compute_col_stats_for_parquet,
@@ -139,6 +142,7 @@ def __getattr__(name):
     "write_success_marker",
     "touch_success_marker",
     "verify_snapshot_integrity",
+    "snapshot_range_checksums",
     "has_resumable_ingest",
     "validated_index_resume_progress",
     "compute_col_stats_for_parquet",

packages/diskgraph/diskgraph-pybind/python/tests/test_exception_exports.py

@@ -11,3 +11,27 @@
 def test_public_corruption_error_catches_snapshot_verifier(tmp_path):
     with pytest.raises(diskgraph.DiskgraphCorruptionError):
         diskgraph.verify_snapshot_integrity(str(tmp_path))
+
+
+def test_build_identity_is_exported():
+    assert diskgraph.__version__
+    assert diskgraph.BUILD_COMMIT
+
+
+def test_snapshot_range_checksums_match_zlib(tmp_path):
+    import json
+    import zlib
+
+    payload = bytes(range(256)) * 5
+    (tmp_path / "a.bin").write_bytes(payload)
+    (tmp_path / "_MANIFEST.json").write_text(json.dumps({
+        "schema_version": 1,
+        "files": [{"rel_path": "a.bin", "size_bytes": len(payload)}],
+    }))
+    crcs = diskgraph.snapshot_range_checksums(str(tmp_path), 512)
+    assert crcs["a.bin"] == [zlib.crc32(payload[i:i + 512])
+                             for i in range(0, len(payload), 512)]
+    assert crcs["_MANIFEST.json"] == [
+        zlib.crc32((tmp_path / "_MANIFEST.json").read_bytes())]
+    with pytest.raises(diskgraph.DiskgraphCorruptionError):
+        diskgraph.snapshot_range_checksums(str(tmp_path / "missing"), 512)

packages/diskgraph/diskgraph-pybind/src/lib.rs

@@ -1553,6 +1553,21 @@ fn verify_snapshot_integrity(py: Python<'_>, output_dir: &str) -> PyResult<()> {
         .map_err(|e| DiskgraphCorruptionError::new_err(e.to_string()))
 }
 
+/// CRC-32 of every `range_bytes` span of each file in the snapshot's
+/// `_MANIFEST.json` and of the manifest itself, keyed by relative path.
+#[pyfunction]
+fn snapshot_range_checksums(
+    py: Python<'_>,
+    output_dir: &str,
+    range_bytes: u64,
+) -> PyResult<std::collections::BTreeMap<String, Vec<u32>>> {
+    let output_dir = std::path::PathBuf::from(output_dir);
+    py.detach(|| {
+        diskgraph_ingest::snapshot_mirror::snapshot_range_checksums(&output_dir, range_bytes)
+    })
+    .map_err(|e| DiskgraphCorruptionError::new_err(e.to_string()))
+}
+
 #[pyfunction]
 fn has_resumable_ingest(output_dir: &str) -> PyResult<bool> {
     diskgraph_ingest::initial_resume_phase(std::path::Path::new(output_dir))
@@ -1593,6 +1608,8 @@ fn diskgraph_pybind(m: &Bound<'_, PyModule>) -> PyResult<()> {
         "DiskgraphCorruptionError",
         m.py().get_type::<DiskgraphCorruptionError>(),
     )?;
+    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
+    m.add("BUILD_COMMIT", env!("DISKGRAPH_BUILD_COMMIT"))?;
     m.add_function(wrap_pyfunction!(export_tt_artifacts, m)?)?;
     m.add_function(wrap_pyfunction!(convert_kumo_config, m)?)?;
     m.add_function(wrap_pyfunction!(ingest, m)?)?;
@@ -1606,6 +1623,7 @@ fn diskgraph_pybind(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(success_marker::write_success_marker, m)?)?;
     m.add_function(wrap_pyfunction!(success_marker::touch_success_marker, m)?)?;
     m.add_function(wrap_pyfunction!(verify_snapshot_integrity, m)?)?;
+    m.add_function(wrap_pyfunction!(snapshot_range_checksums, m)?)?;
     m.add_function(wrap_pyfunction!(has_resumable_ingest, m)?)?;
     m.add_function(wrap_pyfunction!(validated_index_resume_progress, m)?)?;
     m.add_function(wrap_pyfunction!(compute_col_stats_for_parquet, m)?)?;

packages/diskgraph/third_party/artifact-manifest.json

@@ -3241,6 +3241,13 @@
           "source_kind": "registry-or-git",
           "version": "0.3.0"
         },
+        {
+          "license": "MIT OR Apache-2.0",
+          "name": "crc32fast",
+          "source": "registry+https://github.com/rust-lang/crates.io-index",
+          "source_kind": "registry-or-git",
+          "version": "1.5.0"
+        },
         {
           "license": "MIT OR Apache-2.0",
           "name": "crossbeam-deque",
@@ -5202,6 +5209,6 @@
       ]
     }
   ],
-  "cargo_lock_sha256": "297851eb08303a13a97f091c68cf3ce5ccac633ad3d7fe09724c1506e486cdaf",
+  "cargo_lock_sha256": "a1c319a5eb83c1a60f056c213932c12e385f0cff468da4db0021947f11b7aa07",
   "schema": 1
 }