diff --git a/Cargo.lock b/Cargo.lock
index da7c93293..5723d3bdb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -10534,6 +10534,7 @@ dependencies = [
  "arc-swap",
  "assert_matches",
  "async-channel 2.5.0",
+ "async-compression",
  "async-recursion",
  "async-trait",
  "aws-sdk-s3",
@@ -10546,6 +10547,7 @@ dependencies = [
  "chrono",
  "cron",
  "dashmap 5.5.3",
+ "flate2",
  "futures 0.3.31",
  "futures-util",
  "http 0.2.9",
@@ -10578,8 +10580,10 @@ dependencies = [
  "serde_json",
  "serde_urlencoded",
  "serde_with",
+ "serde_yaml 0.9.34+deprecated",
  "smallvec",
  "snafu 0.7.5",
+ "snap",
  "stream-cancel",
  "tempfile",
  "test-case",
@@ -10599,6 +10603,7 @@ dependencies = [
  "vector-core",
  "vector-lib",
  "vrl",
+ "zstd 0.13.2",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 5930cb3aa..a05be1d16 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -178,6 +178,7 @@ rand = { version = "0.8.5", default-features = false, features = ["small_rng"] }
 serde_json = { version = "=1.0.140", default-features = false, features = ["raw_value", "std"] }
 serde = { version = "1.0.215", default-features = false, features = ["alloc", "derive", "rc"] }
 serde_with = { version = "3.12.0", default-features = false, features = ["macros", "std"] }
+serde_yaml = { version = "0.9.34", default-features = false }
 snafu = { version = "0.7.5", default-features = false, features = ["futures", "std"] }
 rusqlite = { version = "0.37.0", features = [ "bundled" ] }
 tokio = { version = "1.43.0", default-features = false, features = ["full"] }
@@ -205,7 +206,7 @@ regex = { version = "1.11.1", default-features = false, features = ["std", "perf
 tokio-util = { version = "0.7", default-features = false, features = ["io", "time"] }
 bytesize = { version = "1.3.0", default-features = false }
 bytes = { version = "1.9.0", default-features = false, features = ["serde"] }
-async-compression = { version = "0.4.18", default-features = false, features = ["tokio", "gzip", "zstd"] }
+async-compression = { version = "0.4.18", default-features = false, features = ["tokio", "gzip", "zstd", "zlib", "deflate"] }
 object_store = { version = "0.10.2", features = ["gcp"], default-features = false }
 once_cell = { version = "1.20.2", default-features = false }
 smallvec = { version = "1", default-features = false, features = ["union", "serde"] }
diff --git a/lib/observo/private b/lib/observo/private
index bbdd4686b..26d3e1507 160000
--- a/lib/observo/private
+++ b/lib/observo/private
@@ -1 +1 @@
-Subproject commit bbdd4686bf16dc530637b621333bcf2cd1106598
+Subproject commit 26d3e1507fcc9911ac398e87a555eb5d0edf120b
diff --git a/lib/observo/scol/Cargo.toml b/lib/observo/scol/Cargo.toml
index 1378c130a..24547e4e1 100644
--- a/lib/observo/scol/Cargo.toml
+++ b/lib/observo/scol/Cargo.toml
@@ -66,6 +66,8 @@ reqwest = { version = "0.12", default-features = false }
 # core). Pull hyper 1 under an alias so retry classification can downcast `object_store::Error`
 # sources to the *matching* `hyper::Error` type (see objstore::is_retriable).
 hyper1 = { package = "hyper", version = "1", default-features = false }
+async-compression = { workspace = true }
+snap.workspace = true
 
 # Optional dependencies for testing (below)
 testcontainers = { version = "0.19", optional = true }
@@ -92,10 +94,13 @@ regex = "1"
 serde_urlencoded = "0.7"
 tempfile = "3.15.0"
 tracing-test = { version = "0.2.5", features = [ "no-env-filter" ] }
+serde_yaml = { workspace = true }
 test-case = { version = "3.3" }
 tokio = { version = "1.43.0", features = ["test-util"] }
 scol = { path = ".", features = ["test-scenarios"] }
 chkpts = { path = "../chkpts" }
+flate2 = { workspace = true }
+zstd = { workspace = true }
 
 [features]
 test-scenarios = ["vector-lib/test", "dep:textwrap", "dep:rstest", "dep:oxide-auth",
diff --git a/lib/vector-common/src/chkpts.rs b/lib/vector-common/src/chkpts.rs
index ab507fdc6..b1db07346 100644
--- a/lib/vector-common/src/chkpts.rs
+++ b/lib/vector-common/src/chkpts.rs
@@ -39,6 +39,24 @@ pub enum ChkptErr {
     Unknown(crate::Error)
 }
 
+/// Precondition guarding a `compare_and_set`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SetCond<'a> {
+    /// The checkpoint must exist and its latest value must equal this.
+    Old(Cow<'a, str>),
+    /// The checkpoint must not exist.
+    Absent,
+}
+
+impl SetCond<'_> {
+    pub fn into_owned(self) -> SetCond<'static> {
+        match self {
+            SetCond::Old(v) => SetCond::Old(Cow::Owned(v.into_owned())),
+            SetCond::Absent => SetCond::Absent,
+        }
+    }
+}
+
 impl Display for ChkptErr {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{self:?}")
@@ -58,7 +76,7 @@ impl Error for ChkptErr {
 pub trait Accessor: Send + dyn_clone::DynClone + Sync {
     async fn get(&self, id: Cow<'_, str>) -> Result<Value, ChkptErr>;
     async fn set(&self, id: Cow<'_, str>, value: Cow<'_, str>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>;
-    async fn compare_and_set(&self, id: Cow<'_, str>, value: Cow<'_, str>, if_old: Cow<'_, str>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>;
+    async fn compare_and_set(&self, id: Cow<'_, str>, value: Cow<'_, str>, cond: SetCond<'_>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>;
     async fn del(&self, id: Cow<'_, str>) -> Result<(), ChkptErr>;
     async fn compare_and_del(&self, id: Cow<'_, str>, if_old: Cow<'_, str>) -> Result<(), ChkptErr>;
     async fn del_range(&self, from: Cow<'_, str>, to: Cow<'_, str>) -> Result<u64, ChkptErr>;
diff --git a/src/sources/scol/mod.rs b/src/sources/scol/mod.rs
index 1e12ef700..e822b59c3 100644
--- a/src/sources/scol/mod.rs
+++ b/src/sources/scol/mod.rs
@@ -23,7 +23,7 @@ impl SourceConfig for Config {
         let chkptr = cx.checkpoint_accessor().await;
         let src = self
             .clone()
-            .build_source(cx.out, cx.shutdown, chkptr, lns)
+            .build_source(cx.out, cx.shutdown, chkptr, lns, cx.globals.limits.compression)
             .map(|r| match r {
                 Ok(_) => Ok(()),
                 Err(e) => {