feat: implement s3 dump extension

Taskfile.yml

@@ -298,12 +298,52 @@ tasks:
       - task: test
       - echo "✅ All checks passed"
 
+  test:s3-e2e:
+    desc: "s3-dump MinIO smoke test: dump a Datadog log to a real S3 server (requires docker)"
+    cmds:
+      - |
+        if ! command -v docker >/dev/null 2>&1; then
+          echo "docker is required for task test:s3-e2e"
+          exit 1
+        fi
+        CONTAINER=edge-minio-e2e
+        NETWORK=edge-minio-e2e-net
+        BUCKET=edge-e2e
+        cleanup() {
+          docker rm -f "$CONTAINER" >/dev/null 2>&1 || true
+          docker network rm "$NETWORK" >/dev/null 2>&1 || true
+        }
+        trap cleanup EXIT
+        cleanup
+        docker network create "$NETWORK" >/dev/null
+        docker run -d --rm --name "$CONTAINER" --network "$NETWORK" \
+          -p 9000:9000 \
+          -e MINIO_ROOT_USER=minioadmin \
+          -e MINIO_ROOT_PASSWORD=minioadmin \
+          minio/minio server /data >/dev/null
+        # Wait for the server to report healthy on the published port.
+        for i in $(seq 1 30); do
+          if curl -fsS http://127.0.0.1:9000/minio/health/live >/dev/null 2>&1; then break; fi
+          if [ "$i" = "30" ]; then echo "MinIO did not become healthy"; docker logs "$CONTAINER" || true; exit 1; fi
+          sleep 1
+        done
+        # Edge has no S3 admin client, so create the bucket with mc on the same
+        # docker network (reaches the server by container name).
+        docker run --rm --network "$NETWORK" --entrypoint /bin/sh minio/mc -c \
+          "mc alias set local http://$CONTAINER:9000 minioadmin minioadmin >/dev/null && mc mb -p local/$BUCKET >/dev/null"
+        AWS_ACCESS_KEY_ID=minioadmin \
+        AWS_SECRET_ACCESS_KEY=minioadmin \
+        S3_ENDPOINT=http://127.0.0.1:9000 \
+        S3_BUCKET="$BUCKET" \
+          zig build test-s3-e2e --summary all
+
   signoff:
     desc: "Sign off the codebase"
     aliases: ["sign", "s"]
     cmds:
       - task: do
       - task: build:safe
+      - task: test:s3-e2e
       - gh signoff
 
   install:

build.zig

@@ -61,6 +61,9 @@ pub fn build(b: *std.Build) void {
     // Shared modules from policy-zig ensure type identity across boundaries.
     const proto_mod = policy_dep.module("proto");
     const o11y_mod = policy_dep.module("observability");
+    // Optional extensions module (pulls in the z3 S3 client). Only edge code
+    // that wires s3-dump imports it.
+    const ext_mod = policy_dep.module("extensions");
 
     // ==========================================================================
     // Edge Library Module
@@ -74,6 +77,7 @@ pub fn build(b: *std.Build) void {
             .{ .name = "zimdjson", .module = zimdjson.module("zimdjson") },
             .{ .name = "policy_zig", .module = policy_dep.module("policy_zig") },
             .{ .name = "o11y", .module = o11y_mod },
+            .{ .name = "extensions", .module = ext_mod },
             .{ .name = "metrics_zig", .module = metrics_dep.module("metrics") },
             .{ .name = "httpz", .module = httpz_mod },
         },
@@ -111,6 +115,7 @@ pub fn build(b: *std.Build) void {
     exe.root_module.addImport("zimdjson", zimdjson.module("zimdjson"));
     exe.root_module.addImport("policy_zig", policy_dep.module("policy_zig"));
     exe.root_module.addImport("o11y", o11y_mod);
+    exe.root_module.addImport("extensions", ext_mod);
     exe.root_module.addImport("metrics_zig", metrics_dep.module("metrics"));
     exe.root_module.addImport("httpz", httpz_mod);
     exe.root_module.addOptions("build_options", build_options);
@@ -152,6 +157,7 @@ pub fn build(b: *std.Build) void {
         dist_exe.root_module.addImport("zimdjson", zimdjson.module("zimdjson"));
         dist_exe.root_module.addImport("policy_zig", policy_dep.module("policy_zig"));
         dist_exe.root_module.addImport("o11y", o11y_mod);
+        dist_exe.root_module.addImport("extensions", ext_mod);
         dist_exe.root_module.addImport("metrics_zig", metrics_dep.module("metrics"));
         dist_exe.root_module.addImport("httpz", httpz_mod);
         dist_exe.root_module.addOptions("build_options", build_options);
@@ -208,6 +214,26 @@ pub fn build(b: *std.Build) void {
     const test_step = b.step("test", "Run tests");
     test_step.dependOn(&run_mod_tests.step);
 
+    // Real-storage smoke test for the s3-dump extension, filtered to the MinIO
+    // e2e test. Excluded from `test` (it needs a live backend); driven by
+    // `task test:s3-e2e`, which starts MinIO, creates the bucket, and sets the
+    // S3 env vars this test reads.
+    const s3_e2e_tests = b.addTest(.{
+        .root_module = mod,
+        .filters = &.{"e2e minio"},
+    });
+    s3_e2e_tests.root_module.link_libc = true;
+    s3_e2e_tests.root_module.linkSystemLibrary("z", .{});
+    s3_e2e_tests.root_module.linkSystemLibrary("zstd", .{});
+    s3_e2e_tests.root_module.addImport("metrics_zig", metrics_dep.module("metrics"));
+    s3_e2e_tests.root_module.addOptions("build_options", build_options);
+    s3_e2e_tests.root_module.addAnonymousImport("otlp_metrics_benchmark_pb", .{
+        .root_source_file = b.path("bench/scaling/payloads/otlp-metrics.pb"),
+    });
+    const run_s3_e2e_tests = b.addRunArtifact(s3_e2e_tests);
+    const s3_e2e_step = b.step("test-s3-e2e", "Run the s3-dump MinIO smoke test (needs S3 env vars)");
+    s3_e2e_step.dependOn(&run_s3_e2e_tests.step);
+
     // ==========================================================================
     // Benchmark Tools
     // ==========================================================================

build.zig.zon

@@ -49,8 +49,8 @@
             .hash = "zbench-0.13.0-YTdc7xVBAQCCMC-IdLLuotBeiNNNm8k9Pi2V4VYqrwfI",
         },
         .policy_zig = .{
-            .url = "https://github.com/usetero/policy-zig/archive/refs/tags/v0.4.4.tar.gz",
-            .hash = "policy_zig-0.1.7-5_dp3q0-EgDyYRR7_LInB9k-CjcBiI8nGewWKOXAVbvO",
+            .url = "https://github.com/usetero/policy-zig/archive/refs/tags/v0.5.0.tar.gz",
+            .hash = "policy_zig-0.1.7-5_dp3j7lFAAWNWhq8DkCv990ujiTWKWjYxzsIKNpeXIT",
         },
     },
     .paths = .{

src/bench/datadog_log_bench.zig

@@ -267,6 +267,7 @@ fn evalOne(registry: *PolicyRegistry, record: []const u8) std.meta.Tag(logs.Reco
         registry,
         g_bus,
         record,
+        null,
     ) catch |err| std.debug.panic("evalLogRecord failed: {t}", .{err});
     return std.meta.activeTag(verdict);
 }

src/config/types.zig

@@ -1,5 +1,8 @@
+const std = @import("std");
 const policy = @import("policy_zig");
 
+const log = std.log.scoped(.config);
+
 pub const ProviderConfig = policy.ProviderConfig;
 pub const ServiceMetadata = policy.ServiceMetadata;
 pub const StringPair = policy.StringPair;
@@ -24,6 +27,36 @@ pub const PrometheusModuleConfig = struct {
     max_output_bytes_per_scrape: usize = 10 * 1024 * 1024,
 };
 
+/// A single S3-compatible destination the s3-dump extension can write to.
+/// Fields (except `name`) mirror the z3 target config the extension expects;
+/// `name` is what policies reference via `ExtensionTargetRef`.
+pub const S3TargetConfig = struct {
+    name: []const u8,
+    /// S3-compatible endpoint URL. Null uses the AWS default for `region`.
+    endpoint: ?[]const u8 = null,
+    region: []const u8 = "us-east-1",
+    bucket: []const u8,
+    prefix: []const u8 = "",
+    /// Path-style (MinIO/R2) vs virtual-host-style URLs.
+    force_path_style: bool = true,
+};
+
+/// `com.usetero/s3-dump` extension configuration. Off unless `enabled` and at
+/// least one target is set. Credentials are NOT here — they come from the
+/// AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY environment variables.
+pub const S3DumpConfig = struct {
+    enabled: bool = false,
+    /// Interval between background flush cycles (single flush task).
+    flush_interval_ms: u64 = 30_000,
+    // Batching knobs — map 1:1 onto the extension's S3Dump.Options.
+    max_batch_bytes: usize = 4 << 20,
+    max_batch_records: u32 = 10_000,
+    max_batch_age_ms: u64 = 30_000,
+    max_sealed_bytes: usize = 32 << 20,
+    max_attempts: usize = 1,
+    targets: []const S3TargetConfig = &.{},
+};
+
 /// Main proxy configuration - loadable via zonfig
 pub const ProxyConfig = struct {
     // Network config
@@ -69,4 +102,42 @@ pub const ProxyConfig = struct {
 
     // Module-specific configuration
     prometheus: PrometheusModuleConfig = .{},
+
+    /// com.usetero/s3-dump extension (Datadog logs → S3). Disabled by default.
+    s3_dump: S3DumpConfig = .{},
+
+    /// Post-load validation hook (called by zonfig). Rejects s3-dump knobs that
+    /// would silently break the runtime: a zero flush interval spins the flush
+    /// task in a tight loop, and a zero batch/backlog cap makes the handler drop
+    /// every record. Only checked when the extension is enabled.
+    pub fn validate(self: *ProxyConfig) !void {
+        const s = self.s3_dump;
+        if (!s.enabled) return;
+        if (s.flush_interval_ms == 0) {
+            log.warn("s3_dump.flush_interval_ms must be > 0", .{});
+            return error.InvalidS3DumpConfig;
+        }
+        if (s.max_batch_bytes == 0 or s.max_sealed_bytes == 0 or s.max_batch_records == 0) {
+            log.warn("s3_dump batch caps (max_batch_bytes/max_sealed_bytes/max_batch_records) must be > 0", .{});
+            return error.InvalidS3DumpConfig;
+        }
+    }
 };
+
+test "ProxyConfig.validate rejects zero s3_dump knobs when enabled" {
+    // Disabled: knobs are ignored even if degenerate.
+    var disabled: ProxyConfig = .{ .s3_dump = .{ .enabled = false, .flush_interval_ms = 0 } };
+    try disabled.validate();
+
+    // Enabled with a zero flush interval → tight-spin guard.
+    var zero_interval: ProxyConfig = .{ .s3_dump = .{ .enabled = true, .flush_interval_ms = 0 } };
+    try std.testing.expectError(error.InvalidS3DumpConfig, zero_interval.validate());
+
+    // Enabled with a zero batch cap → every record would drop.
+    var zero_bytes: ProxyConfig = .{ .s3_dump = .{ .enabled = true, .max_batch_bytes = 0 } };
+    try std.testing.expectError(error.InvalidS3DumpConfig, zero_bytes.validate());
+
+    // Enabled with sane defaults → accepted.
+    var ok: ProxyConfig = .{ .s3_dump = .{ .enabled = true } };
+    try ok.validate();
+}

src/frontend/exec.zig

@@ -68,6 +68,9 @@ pub const SharedCtx = struct {
     limits: limits_mod.Limits,
     /// Debug tap, or null when disabled by config. See `TapState`.
     tap: ?*TapState = null,
+    /// Extension dispatch sink (s3-dump), or null when extensions are off.
+    /// Threaded into per-record policy evaluation on the Datadog log path.
+    extension_sink: ?policy.ExtensionSink = null,
 };
 
 /// Routes and plans a request from transport-neutral parts. Returns null
@@ -411,6 +414,7 @@ pub const RecordSink = struct {
             self.ctx.registry,
             self.ctx.bus,
             bytes,
+            self.ctx.extension_sink,
         );
         switch (verdict) {
             .keep => return .keep,

src/root.zig

@@ -100,6 +100,7 @@ test {
     _ = @import("frontend/stdio/server.zig");
     _ = @import("frontend/httpz/server.zig");
     _ = @import("runtime/distro.zig");
+    _ = @import("runtime/extensions.zig");
     _ = @import("signals/prometheus/root.zig");
     _ = @import("lambda/root.zig");
     _ = @import("zonfig/root.zig");

src/runtime/app.zig

@@ -22,6 +22,7 @@ const router_mod = @import("../service/router.zig");
 const upstream_mod = @import("../frontend/upstream.zig");
 const exec_mod = @import("../frontend/exec.zig");
 const frontend_select = @import("../frontend/select.zig");
+const ext_rt = @import("extensions.zig");
 
 const policy = @import("policy_zig");
 const o11y = @import("o11y");
@@ -223,6 +224,7 @@ pub const EngineOptions = struct {
     metrics_url: ?[]const u8 = null,
     service_options: distro.ServiceOptions = .{},
     tap_enabled: bool = false,
+    extension_sink: ?policy.ExtensionSink = null,
 };
 
 pub const Engine = struct {
@@ -310,6 +312,7 @@ pub const Engine = struct {
             .metrics = metrics,
             .limits = self.limits,
             .tap = if (options.tap_enabled) &self.tap else null,
+            .extension_sink = options.extension_sink,
         };
 
         self.server = try .init(
@@ -449,6 +452,18 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
     var registry = policy.Registry.init(allocator, bus);
     defer registry.deinit();
 
+    // s3-dump extension: wired before the loader subscribes providers so the
+    // sync hooks reach every provider. Stays inert unless enabled with targets.
+    var exts = ext_rt.Extensions.init(allocator, .{ .log = ext_rt.datadogLogEncode });
+    defer exts.deinit();
+    const s3_dump_active = config.s3_dump.enabled and config.s3_dump.targets.len > 0;
+    var extension_sink: ?policy.ExtensionSink = null;
+    if (s3_dump_active) {
+        try ext_rt.configure(&exts, allocator, io, config.s3_dump, init.environ_map);
+        exts.register(&registry);
+        extension_sink = exts.sink();
+    }
+
     var runtime_metrics = try RuntimeMetrics.init(allocator, io, distributionLabel(distribution));
     defer runtime_metrics.deinit();
     runtime_metrics.setBuildInfo(build_options.version, build_options.commit);
@@ -475,6 +490,7 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
             .prometheus_max_output_bytes = config.prometheus.max_output_bytes_per_scrape,
         },
         .tap_enabled = config.tap_enabled,
+        .extension_sink = extension_sink,
     });
     defer engine.destroy();
 
@@ -490,6 +506,17 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
 
     try engine.start();
 
+    // Single background flush task, torn down with the rest of the group on
+    // shutdown; the final force-flush runs on the main path below.
+    if (s3_dump_active) {
+        try engine.lifecycle.spawn(io, ext_rt.flushLoop, .{
+            &exts,
+            io,
+            config.s3_dump.flush_interval_ms,
+            &runtime_metrics,
+        });
+    }
+
     // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
     bus.info(DataPlaneBudget{
         .max_connections = engine.limits.max_connections,
@@ -506,6 +533,11 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
     engine.awaitShutdown();
     engine.stop();
 
+    // Drain any buffered dump batches before we tear down (io is still live).
+    if (s3_dump_active) {
+        runtime_metrics.recordS3DumpFlush(exts.flush(io, .{ .force = true }));
+    }
+
     if (signal_waiter) |waiter| {
         shutdown_waiter.store(true, .release);
         std.posix.kill(std.c.getpid(), std.posix.SIG.USR1) catch |err|