feat: enable lambda s3 dumps

src/config/types.zig

@@ -55,6 +55,30 @@ pub const S3DumpConfig = struct {
     max_sealed_bytes: usize = 32 << 20,
     max_attempts: usize = 1,
     targets: []const S3TargetConfig = &.{},
+
+    /// Batch/backlog caps that must be nonzero no matter how flush is driven —
+    /// a zero cap makes the handler treat every record as oversized and drop it.
+    /// Used by every distribution (the Lambda flush is event-driven, so it only
+    /// needs this, not the flush-interval check below).
+    pub fn validateBatching(self: S3DumpConfig) !void {
+        if (!self.enabled) return;
+        if (self.max_batch_bytes == 0 or self.max_sealed_bytes == 0 or self.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;
+        }
+    }
+
+    /// Full validation for distributions that run the wall-clock flush task
+    /// (the server distro): the batch caps plus a nonzero flush interval, since
+    /// a zero interval spins the flush loop tight. Lambda calls
+    /// `validateBatching` directly instead — `flush_interval_ms` is unused there.
+    pub fn validate(self: S3DumpConfig) !void {
+        try self.validateBatching();
+        if (self.enabled and self.flush_interval_ms == 0) {
+            log.warn("s3_dump.flush_interval_ms must be > 0", .{});
+            return error.InvalidS3DumpConfig;
+        }
+    }
 };
 
 /// Main proxy configuration - loadable via zonfig
@@ -106,21 +130,9 @@ pub const ProxyConfig = struct {
     /// 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.
+    /// Post-load validation hook (called by zonfig).
     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;
-        }
+        try self.s3_dump.validate();
     }
 };
 
@@ -141,3 +153,15 @@ test "ProxyConfig.validate rejects zero s3_dump knobs when enabled" {
     var ok: ProxyConfig = .{ .s3_dump = .{ .enabled = true } };
     try ok.validate();
 }
+
+test "S3DumpConfig.validateBatching (Lambda path) ignores flush_interval_ms" {
+    // Lambda flush is event-driven, so a zero interval is fine there...
+    const zero_interval: S3DumpConfig = .{ .enabled = true, .flush_interval_ms = 0 };
+    try zero_interval.validateBatching();
+    // ...but the full validate (server distro) still rejects it.
+    try std.testing.expectError(error.InvalidS3DumpConfig, zero_interval.validate());
+
+    // Zero batch caps are rejected on both paths.
+    const zero_cap: S3DumpConfig = .{ .enabled = true, .max_batch_bytes = 0 };
+    try std.testing.expectError(error.InvalidS3DumpConfig, zero_cap.validateBatching());
+}

src/lambda_main.zig

@@ -19,6 +19,8 @@ const build_options = @import("build_options");
 const edge = @import("root.zig");
 const runtime_metrics_mod = @import("runtime/runtime_metrics.zig");
 const app = @import("runtime/app.zig");
+const ext_rt = @import("runtime/extensions.zig");
+const config_types = @import("config/types.zig");
 const policy = edge.policy;
 const zonfig = edge.zonfig;
 
@@ -91,6 +93,11 @@ pub const LambdaConfig = struct {
         api_key: ?[]const u8 = null,
     } = .{},
 
+    /// com.usetero/s3-dump extension (Datadog logs → S3). Disabled by default.
+    /// In Lambda the flush is event-driven (per-invoke + shutdown), so
+    /// `flush_interval_ms` is unused here — see the event loop in `main`.
+    s3_dump: config_types.S3DumpConfig = .{},
+
     /// zonfig validation hook, run after env/JSON overrides. Clamps env knobs
     /// into ranges the runtime requires so a bad override degrades instead of
     /// aborting (or silently wedging) the extension at startup.
@@ -118,6 +125,10 @@ pub const LambdaConfig = struct {
             std.log.warn("TERO_THREAD_POOL_COUNT=0 starts no handler threads; using httpz default", .{});
             self.thread_pool_count = null;
         };
+
+        // Lambda flush is event-driven (per-invoke + shutdown), so
+        // `flush_interval_ms` is unused here — validate only the batch caps.
+        try self.s3_dump.validateBatching();
     }
 };
 
@@ -227,6 +238,18 @@ pub fn main(init: std.process.Init) !void {
     var registry = policy.Registry.init(allocator, bus);
     defer registry.deinit();
 
+    // s3-dump extension: wired before policies compile so extension bindings
+    // resolve at snapshot build. 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, .lambda);
     defer runtime_metrics.deinit();
     runtime_metrics.setBuildInfo(build_options.version, build_options.commit);
@@ -305,6 +328,7 @@ pub fn main(init: std.process.Init) !void {
         .upstream_url = config.upstream_url,
         .logs_url = config.logs_url,
         .metrics_url = config.metrics_url,
+        .extension_sink = extension_sink,
     });
     defer engine.destroy();
 
@@ -342,14 +366,22 @@ pub fn main(init: std.process.Init) !void {
                 bus.debug(LambdaInvokeReceived{
                     .request_id = invoke_event.request_id,
                 });
-                // Proxy server handles requests in background
-                // Nothing to do here - just continue polling
+                // Lambda freezes the environment between invocations, so the
+                // server distro's wall-clock flush timer is unreliable here.
+                // Flush at each invoke boundary instead: receiving this event
+                // means the PRIOR invocation's function has completed, so its
+                // telemetry is fully batched. The final invocation's tail is
+                // drained by the SHUTDOWN force-flush below.
+                if (s3_dump_active) runtime_metrics.recordS3DumpFlush(exts.flush(io, .{}));
             },
             .shutdown => |shutdown_event| {
                 bus.info(LambdaExtensionShutdown{
                     .reason = @tagName(shutdown_event.reason),
                 });
 
+                // Final drain within Lambda's shutdown deadline (io still live).
+                if (s3_dump_active) runtime_metrics.recordS3DumpFlush(exts.flush(io, .{ .force = true }));
+
                 // Stop proxy server gracefully
                 shutdown_requested.store(true, .release);
                 engine.requestShutdown();

src/runtime/extensions.zig

@@ -43,9 +43,33 @@ pub fn datadogLogEncode(record: *const anyopaque, writer: *std.Io.Writer) anyerr
 /// Read S3 credentials from the environment. Returns null when either half is
 /// absent — the extension then counts-and-drops deliveries for its targets
 /// (fail-open), so a missing credential never stalls telemetry.
+///
+/// The `TERO_S3_*` pair is tried first, then the standard `AWS_*` pair — each
+/// resolved as a UNIT (both halves from the same set), never mixed. A partial
+/// `TERO_S3_*` (id without secret) means the override is intentionally absent,
+/// so we fall through to `AWS_*` rather than pairing keys across credential
+/// sets (which would sign every request with an invalid keypair).
+///
+/// This matters in Lambda: `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY` are
+/// *reserved* — the runtime overwrites them with the execution role's TEMPORARY
+/// creds (which z3 can't sign, since it emits no `x-amz-security-token`). So to
+/// use static IAM-user keys there, set them under the non-reserved `TERO_S3_*`
+/// names, which the runtime never touches. Outside Lambda the `AWS_*` fallback
+/// works as usual.
 pub fn credentialsFromEnv(env: *const std.process.Environ.Map) ?ext.S3Dump.Credentials {
-    const access = env.get("AWS_ACCESS_KEY_ID") orelse return null;
-    const secret = env.get("AWS_SECRET_ACCESS_KEY") orelse return null;
+    return pairFromEnv(env, "TERO_S3_ACCESS_KEY_ID", "TERO_S3_SECRET_ACCESS_KEY") orelse
+        pairFromEnv(env, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY");
+}
+
+/// Resolve a same-set access-key/secret pair, or null if either half is
+/// missing or empty (so the caller can cleanly fall back to the next set).
+fn pairFromEnv(
+    env: *const std.process.Environ.Map,
+    access_name: []const u8,
+    secret_name: []const u8,
+) ?ext.S3Dump.Credentials {
+    const access = env.get(access_name) orelse return null;
+    const secret = env.get(secret_name) orelse return null;
     if (access.len == 0 or secret.len == 0) return null;
     return .{ .access_key_id = access, .secret_access_key = secret };
 }
@@ -107,6 +131,30 @@ pub fn flushLoop(
     }
 }
 
+test "credentialsFromEnv prefers TERO_S3_* over the reserved AWS_* names" {
+    var env = std.process.Environ.Map.init(std.testing.allocator);
+    defer env.deinit();
+
+    // Neither set → null (fail-open: deliveries drop, nothing stalls).
+    try std.testing.expect(credentialsFromEnv(&env) == null);
+
+    // Only AWS_* (in Lambda these are the reserved role creds) → fallback.
+    try env.put("AWS_ACCESS_KEY_ID", "AKIA-fallback");
+    try env.put("AWS_SECRET_ACCESS_KEY", "fallback-secret");
+    try std.testing.expectEqualStrings("AKIA-fallback", credentialsFromEnv(&env).?.access_key_id);
+
+    // A PARTIAL TERO_S3_* (id only) must NOT pair with AWS_*'s secret — that
+    // would sign with a mixed keypair. Fall back to the whole AWS_* pair.
+    try env.put("TERO_S3_ACCESS_KEY_ID", "AKIA-user");
+    try std.testing.expectEqualStrings("AKIA-fallback", credentialsFromEnv(&env).?.access_key_id);
+
+    // Complete TERO_S3_* pair → wins (the Lambda static-key path).
+    try env.put("TERO_S3_SECRET_ACCESS_KEY", "user-secret");
+    const creds = credentialsFromEnv(&env).?;
+    try std.testing.expectEqualStrings("AKIA-user", creds.access_key_id);
+    try std.testing.expectEqualStrings("user-secret", creds.secret_access_key);
+}
+
 test "datadogLogEncode renders the delivered log context as JSON" {
     const allocator = std.testing.allocator;
     var ddlog = try datadog_log.DatadogLog.parseRaw(allocator, "{\"status\":\"info\",\"message\":\"hi\"}");