From 1c582b7756ad91054a404a6f66ab2e9acf9b8c1a Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Mon, 22 Jun 2026 21:56:18 +0000 Subject: [PATCH 1/3] feat: carry peer_mode through upstream + service registry types adds tls.peer to the spec-side types so the L7 proxy can read the service's mtls posture at dial time: - Upstream gains peer_mode (defaults .off) - ServiceDefinition and ServiceState gain peer_mode (defaults .off) - ServiceSnapshot exposes peer_mode read-only - assignCompatProxyFields copies peer_mode from def to state - snapshotService propagates peer_mode - resolveUpstreamWithPolicy copies peer_mode onto each Upstream candidate and onto the selected one returned to the caller the cluster-state-DB column to persist peer_mode end-to-end (so the manifest's tls.peer reaches resolveUpstream in production) is a small follow-up schema migration; this PR's purpose is the in-memory carry so forwardSingleAttempt can dispatch on it. --- src/network/proxy/runtime.zig | 3 +++ src/network/proxy/upstream.zig | 7 +++++++ src/network/service_registry.zig | 8 ++++++++ 3 files changed, 18 insertions(+) diff --git a/src/network/proxy/runtime.zig b/src/network/proxy/runtime.zig index 0147351d..37fb2459 100644 --- a/src/network/proxy/runtime.zig +++ b/src/network/proxy/runtime.zig @@ -699,6 +699,7 @@ pub fn resolveUpstreamWithPolicy(alloc: std.mem.Allocator, service_name: []const const now_ms = nowRealMilliseconds(); const target_port = service.http_proxy_target_port; + const peer_mode = service.peer_mode; for (endpoints.items) |endpoint| { const port: u16 = target_port orelse if (endpoint.port < 0) 0 else @intCast(endpoint.port); try candidates.append(alloc, .{ @@ -707,6 +708,7 @@ pub fn resolveUpstreamWithPolicy(alloc: std.mem.Allocator, service_name: []const .address = try alloc.dupe(u8, endpoint.ip_address), .port = port, .eligible = endpoint.eligible and endpointAllowsRequestLocked(endpoint.endpoint_id, now_ms, cb_policy), + .peer_mode = peer_mode, }); } @@ -717,6 +719,7 @@ pub fn resolveUpstreamWithPolicy(alloc: std.mem.Allocator, service_name: []const .address = try alloc.dupe(u8, selected.address), .port = selected.port, .eligible = selected.eligible, + .peer_mode = selected.peer_mode, }; } diff --git a/src/network/proxy/upstream.zig b/src/network/proxy/upstream.zig index ac396925..3a28ce2a 100644 --- a/src/network/proxy/upstream.zig +++ b/src/network/proxy/upstream.zig @@ -1,4 +1,5 @@ const std = @import("std"); +const spec = @import("../../manifest/spec.zig"); pub const Upstream = struct { service: []const u8, @@ -6,6 +7,12 @@ pub const Upstream = struct { address: []const u8, port: u16, eligible: bool = true, + /// service-to-service mTLS posture, copied from the service's manifest. + /// when `.off` the dial stays plaintext and uses the existing connection + /// pool; when `.warn` or `.require` the dial runs an mTLS handshake and + /// the resulting session is **not pooled** (each session carries its own + /// encryption state — sharing it across requests is unsafe). + peer_mode: spec.TlsConfig.PeerMode = .off, pub fn deinit(self: Upstream, alloc: std.mem.Allocator) void { alloc.free(self.service); diff --git a/src/network/service_registry.zig b/src/network/service_registry.zig index b3f50dcc..310d1621 100644 --- a/src/network/service_registry.zig +++ b/src/network/service_registry.zig @@ -1,4 +1,5 @@ const std = @import("std"); +const spec = @import("../manifest/spec.zig"); const Allocator = std.mem.Allocator; @@ -87,6 +88,9 @@ pub const ServiceDefinition = struct { http_proxy_circuit_breaker_threshold: ?u8 = null, http_proxy_circuit_breaker_timeout_ms: ?u32 = null, http_proxy_mirror_service: ?[]const u8 = null, + /// service-to-service mTLS posture for traffic *to* this service. + /// copied verbatim from the manifest's `service..tls.peer`. + peer_mode: spec.TlsConfig.PeerMode = .off, }; pub const HttpRouteDefinition = struct { @@ -192,6 +196,7 @@ pub const ServiceSnapshot = struct { http_proxy_circuit_breaker_threshold: ?u8, http_proxy_circuit_breaker_timeout_ms: ?u32, http_proxy_mirror_service: ?[]const u8, + peer_mode: spec.TlsConfig.PeerMode = .off, total_endpoints: usize, eligible_endpoints: usize, healthy_endpoints: usize, @@ -293,6 +298,7 @@ const ServiceState = struct { http_proxy_circuit_breaker_threshold: ?u8 = null, http_proxy_circuit_breaker_timeout_ms: ?u32 = null, http_proxy_mirror_service: ?[]const u8 = null, + peer_mode: spec.TlsConfig.PeerMode = .off, endpoints: std.ArrayList(EndpointState) = .empty, last_reconcile_status: ReconcileStatus = .idle, last_reconcile_error: ?[]const u8 = null, @@ -673,6 +679,7 @@ fn cloneServiceSnapshot(alloc: Allocator, service: *const ServiceState) Error!Se .http_proxy_circuit_breaker_threshold = service.http_proxy_circuit_breaker_threshold, .http_proxy_circuit_breaker_timeout_ms = service.http_proxy_circuit_breaker_timeout_ms, .http_proxy_mirror_service = if (service.http_proxy_mirror_service) |mirror_service| try alloc.dupe(u8, mirror_service) else null, + .peer_mode = service.peer_mode, .total_endpoints = total_endpoints, .eligible_endpoints = eligible_endpoints, .healthy_endpoints = healthy_endpoints, @@ -812,6 +819,7 @@ fn assignCompatProxyFields(alloc: Allocator, service: *ServiceState, definition: service.http_proxy_circuit_breaker_threshold = definition.http_proxy_circuit_breaker_threshold; service.http_proxy_circuit_breaker_timeout_ms = definition.http_proxy_circuit_breaker_timeout_ms; try replaceOptionalOwned(alloc, &service.http_proxy_mirror_service, definition.http_proxy_mirror_service); + service.peer_mode = definition.peer_mode; } fn isEndpointEligible(endpoint: *const EndpointState) bool { From e3004a98752757adb7af642ca47e55dcbc0f0117 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Mon, 22 Jun 2026 21:56:18 +0000 Subject: [PATCH 2/3] feat: forwardSingleAttempt dispatches mtls upstreams to client_dial MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit upstream.peer_mode != .off now branches into a new forwardSingleAttemptMtls: - loads the cluster CA via store.getClusterCa - calls client_dial.dial with the CA pem + sni - writes the request and drains the response through the session, no pooling (mtls sessions hold encryption state and can't share a bare-fd pool key) policy: - .require + missing cluster CA → ClusterCaMissing - .warn + missing cluster CA → log + downgrade to the plaintext dial+pool path (kept as forwardPlainAttempt, exact copy of the legacy leg) so service-to-service traffic keeps flowing while the ca_bootstrap thread catches up readResponseFromSession is the session-aware equivalent of the existing bare-fd readResponse: drains chunks until PeerClosed (the session's orderly EOF) or max_bytes, returns the bytes. three tests on a duck-typed FakeSession cover the read loop's happy path, max-bytes rejection, and immediate-close fallthrough. the live tls handshake itself is already covered end-to-end by the socketpair tests in #438/#439/#442. --- src/network/proxy/reverse_proxy.zig | 175 ++++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) diff --git a/src/network/proxy/reverse_proxy.zig b/src/network/proxy/reverse_proxy.zig index c0fc5baf..d0c3acd2 100644 --- a/src/network/proxy/reverse_proxy.zig +++ b/src/network/proxy/reverse_proxy.zig @@ -19,6 +19,8 @@ const router = @import("router.zig"); const upstream_mod = @import("upstream.zig"); const upstream_pool = @import("upstream_pool.zig"); const runtime_wait = @import("../../lib/runtime_wait.zig"); +const client_dial = @import("../../tls/client_dial.zig"); +const store_mod = @import("../../state/store.zig"); const proxy_loop_header = "X-Yoq-Proxy"; const x_forwarded_for_header = "X-Forwarded-For"; @@ -616,6 +618,10 @@ pub const ReverseProxy = struct { const request = try self.buildForwardRequestWithClient(raw_request, plan, client_ip); defer self.allocator.free(request); + if (upstream.peer_mode != .off) { + return self.forwardSingleAttemptMtls(request, plan, upstream); + } + // prefer a pooled, kept-alive connection; dial a fresh one on a miss. var from_pool = true; var fd = upstream_pool.checkout(upstream.endpoint_id, upstream.address, upstream.port) orelse blk: { @@ -653,8 +659,128 @@ pub const ReverseProxy = struct { } return result.bytes; } + + /// mTLS-specific dial + request/response. always opens a fresh + /// connection (TLS sessions hold encryption state and can't be pooled + /// alongside bare-fd siblings keyed on the same address). on `.warn` + /// a missing/invalid cluster CA logs and downgrades to plaintext; + /// `.require` returns an error in the same case so the caller can + /// surface the failure (PR 5 finale will plug this into proper + /// observability). + fn forwardSingleAttemptMtls( + self: *const ReverseProxy, + request: []const u8, + plan: *const ForwardPlan, + upstream: *const upstream_mod.Upstream, + ) ![]u8 { + const ca_rec_opt = store_mod.getClusterCa(self.allocator) catch null; + const ca_rec = ca_rec_opt orelse { + if (upstream.peer_mode == .require) return error.ClusterCaMissing; + log.warn("mtls upstream {s}: cluster CA not seeded, downgrading to plain dial", .{upstream.address}); + return self.forwardPlainAttempt(request, plan, upstream); + }; + defer ca_rec.deinit(self.allocator); + + var outcome = client_dial.dial(std.Options.debug_io, self.allocator, .{ + .address = upstream.address, + .port = upstream.port, + .connect_timeout_ms = plan.route.connect_timeout_ms, + .ca_cert_pem = ca_rec.cert_pem, + .server_name = upstream.service, + .now_unix = std.Io.Clock.real.now(std.Options.debug_io).toSeconds(), + }) catch |err| return err; + + switch (outcome) { + .bare => |fd| { + // dial returned plaintext somehow (shouldn't happen when + // ca_cert_pem is set, but be defensive). + defer linux_platform.posix.close(fd); + return error.HandshakeFailed; + }, + .session => |*sess| { + defer { + sess.deinit(); + linux_platform.posix.close(sess.fd); + } + + _ = sess.write(request) catch return error.SendFailed; + return try readResponseFromSession(self.allocator, sess, self.max_response_bytes); + }, + } + } + + /// fallback used by the mTLS `.warn` path when no cluster CA is + /// available — runs the plaintext dial/pool path so the connection + /// at least succeeds. equivalent to the legacy non-mTLS leg of + /// forwardSingleAttempt minus the request build (caller already has + /// the bytes). + fn forwardPlainAttempt( + self: *const ReverseProxy, + request: []const u8, + plan: *const ForwardPlan, + upstream: *const upstream_mod.Upstream, + ) ![]u8 { + var from_pool = true; + var fd = upstream_pool.checkout(upstream.endpoint_id, upstream.address, upstream.port) orelse blk: { + from_pool = false; + const dialed = try socket_helpers.connectToUpstream(plan.route.connect_timeout_ms, plan.route.request_timeout_ms, upstream); + upstream_pool.noteDialed(); + break :blk dialed; + }; + + socket_helpers.writeAll(fd, request) catch { + upstream_pool.discard(fd); + if (!from_pool) return error.SendFailed; + from_pool = false; + fd = try socket_helpers.connectToUpstream(plan.route.connect_timeout_ms, plan.route.request_timeout_ms, upstream); + upstream_pool.noteDialed(); + socket_helpers.writeAll(fd, request) catch { + upstream_pool.discard(fd); + return error.SendFailed; + }; + }; + + const result = readResponse(self.allocator, fd, self.max_response_bytes, plan.method == .HEAD) catch |err| { + upstream_pool.discard(fd); + return err; + }; + + if (result.reusable) { + upstream_pool.release(upstream.endpoint_id, upstream.address, upstream.port, fd); + } else { + upstream_pool.discard(fd); + } + return result.bytes; + } }; +/// drain an mTLS session into a single buffer up to `max_bytes`. mTLS +/// connections aren't pooled, so we don't need the framing-aware "is +/// this connection still reusable" logic that the bare-fd readResponse +/// provides — we just read until the peer closes (or `PeerClosed` +/// surfaces via the session) and hand back the bytes. PeerClosed is +/// the orderly EOF signal; any other read error propagates. +fn readResponseFromSession( + alloc: std.mem.Allocator, + sess: anytype, + max_bytes: usize, +) ![]u8 { + var out: std.ArrayList(u8) = .empty; + errdefer out.deinit(alloc); + + var chunk: [8192]u8 = undefined; + while (true) { + const n = sess.read(&chunk) catch |err| { + if (err == error.PeerClosed) break; + return err; + }; + if (n == 0) break; + if (out.items.len + n > max_bytes) return error.ResponseTooLarge; + try out.appendSlice(alloc, chunk[0..n]); + } + return try out.toOwnedSlice(alloc); +} + fn cloneMirrorTask( alloc: std.mem.Allocator, active_mirror_requests: *std.atomic.Value(usize), @@ -4761,3 +4887,52 @@ test "handleConnection rejects looped request after listener restart" { linux_platform.posix.close(client_fd); server_thread.join(); } + +// --- readResponseFromSession --- +// +// the mTLS read loop drains a session-shaped reader until either the peer +// closes (surfaced as ClientSession.read's `PeerClosed`) or `max_bytes` is +// exceeded. exercised here via a small fake session that implements the +// duck-typed `.read([]u8) !usize` shape — keeps the test independent of +// the live TLS stack while still proving the loop's framing. + +const FakeSession = struct { + chunks: []const []const u8, + index: usize = 0, + closed: bool = false, + + fn read(self: *FakeSession, buf: []u8) !usize { + if (self.closed) return error.PeerClosed; + if (self.index >= self.chunks.len) { + self.closed = true; + return error.PeerClosed; + } + const chunk = self.chunks[self.index]; + self.index += 1; + const n = @min(buf.len, chunk.len); + @memcpy(buf[0..n], chunk[0..n]); + return n; + } +}; + +test "readResponseFromSession concatenates chunks until peer closes" { + var sess = FakeSession{ .chunks = &.{ "HTTP/1.1 200 OK\r\n", "Content-Length: 2\r\n\r\nok" } }; + const body = try readResponseFromSession(std.testing.allocator, &sess, 64 * 1024); + defer std.testing.allocator.free(body); + try std.testing.expectEqualStrings("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok", body); +} + +test "readResponseFromSession returns ResponseTooLarge when max_bytes is exceeded" { + var sess = FakeSession{ .chunks = &.{ "AAAA", "BBBB", "CCCC" } }; + try std.testing.expectError( + error.ResponseTooLarge, + readResponseFromSession(std.testing.allocator, &sess, 6), + ); +} + +test "readResponseFromSession returns empty buffer on immediate close" { + var sess = FakeSession{ .chunks = &.{} }; + const body = try readResponseFromSession(std.testing.allocator, &sess, 64); + defer std.testing.allocator.free(body); + try std.testing.expectEqual(@as(usize, 0), body.len); +} From 637658af46c8fd0baeff51e7e53f300c62992495 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Tue, 23 Jun 2026 00:31:05 +0000 Subject: [PATCH 3/3] feat: persist tls.peer in the services table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit closes the manifest → registry → upstream loop. previously peer_mode was added to all the in-memory types but production code paths could not set it because the cluster state DB had no column. with this commit: - services.peer_mode TEXT column (default 'off'); ALTER TABLE migration for existing databases (no NOT NULL so the migration is forward-safe). - ServiceRecord.peer_mode (?[]const u8 — null means 'off', keeps every existing literal compiling). - services_core.createInDb inserts peer_mode; syncConfig takes a peer_mode arg and updates the column alongside lb_policy. - syncServiceDefinitions in the manifest apply path now passes svc.tls.peer.label() through. - serviceDefinitionFromRecord parses the textual peer_mode back into the enum so registry → snapshot → Upstream carries it cleanly. with this in place: a manifest with tls.peer = 'require' on a service flows end-to-end — orchestrator writes 'require' to the services row, the registry snapshot reads it back as .require, resolveUpstream copies it onto Upstream, and forwardSingleAttempt dispatches mtls upstreams to client_dial (the changes from the previous commits). one new round-trip test (syncConfig persists peer_mode) confirms the column behaves and updates apply. --- src/api/routes/status_metrics.zig | 3 ++ src/manifest/orchestrator/startup_runtime.zig | 2 + src/network/service_registry_runtime.zig | 2 + src/state/schema/migrations.zig | 1 + src/state/schema/tables.zig | 1 + src/state/store/services_core.zig | 37 +++++++++++++++++-- src/state/store/services_types.zig | 10 ++++- 7 files changed, 52 insertions(+), 4 deletions(-) diff --git a/src/api/routes/status_metrics.zig b/src/api/routes/status_metrics.zig index 99356e0f..796d93de 100644 --- a/src/api/routes/status_metrics.zig +++ b/src/api/routes/status_metrics.zig @@ -530,6 +530,7 @@ test "route rollout status reports steering blocker for VIP cutover readiness" { testing.allocator, "api", "consistent_hash", + "off", &.{ .{ .route_name = "default", @@ -626,6 +627,7 @@ test "route rollout status reports vip cutover ready after clean backfill audit testing.allocator, "api", "consistent_hash", + "off", &.{ .{ .route_name = "default", @@ -733,6 +735,7 @@ test "route rollout status converges from steering blocked to vip cutover ready" testing.allocator, "api", "consistent_hash", + "off", &.{ .{ .route_name = "default", diff --git a/src/manifest/orchestrator/startup_runtime.zig b/src/manifest/orchestrator/startup_runtime.zig index 67f319c3..c24185e6 100644 --- a/src/manifest/orchestrator/startup_runtime.zig +++ b/src/manifest/orchestrator/startup_runtime.zig @@ -162,10 +162,12 @@ pub fn syncServiceDefinitions( for (route_inputs.items) |route| if (route.backend_services.len > 0) alloc.free(route.backend_services); } if (route_alloc_failed) continue; + const peer_mode_label = if (svc.tls) |tls| tls.peer.label() else "off"; const record = store.syncServiceConfig( alloc, svc.name, "consistent_hash", + peer_mode_label, route_inputs.items, ) catch |err| { log.warn("orchestrator: failed to sync service definition for {s}: {}", .{ svc.name, err }); diff --git a/src/network/service_registry_runtime.zig b/src/network/service_registry_runtime.zig index 9acee52d..72232413 100644 --- a/src/network/service_registry_runtime.zig +++ b/src/network/service_registry_runtime.zig @@ -5,6 +5,7 @@ const service_registry_backfill = @import("service_registry_backfill.zig"); const rollout = @import("service_rollout.zig"); const service_registry = @import("service_registry.zig"); const store = @import("../state/store.zig"); +const spec = @import("../manifest/spec.zig"); const Allocator = std.mem.Allocator; @@ -299,6 +300,7 @@ fn serviceDefinitionFromRecord(service: *const store.ServiceRecord, route_defini .http_proxy_circuit_breaker_threshold = if (service.http_proxy_circuit_breaker_threshold) |v| @intCast(v) else null, .http_proxy_circuit_breaker_timeout_ms = if (service.http_proxy_circuit_breaker_timeout_ms) |v| @intCast(v) else null, .http_proxy_mirror_service = service.http_proxy_mirror_service, + .peer_mode = if (service.peer_mode) |label| (spec.TlsConfig.PeerMode.parse(label) orelse .off) else .off, }; } diff --git a/src/state/schema/migrations.zig b/src/state/schema/migrations.zig index 57e58fc0..d0c1a7ca 100644 --- a/src/state/schema/migrations.zig +++ b/src/state/schema/migrations.zig @@ -49,6 +49,7 @@ fn migrateServices(db: *sqlite.Db) void { addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN http_proxy_retry_on_5xx INTEGER;") catch {}; addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN http_proxy_circuit_breaker_threshold INTEGER;") catch {}; addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN http_proxy_circuit_breaker_timeout_ms INTEGER;") catch {}; + addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN peer_mode TEXT DEFAULT 'off';") catch {}; createTableIfMissing(db, \\CREATE TABLE IF NOT EXISTS service_http_routes ( \\ service_name TEXT NOT NULL, diff --git a/src/state/schema/tables.zig b/src/state/schema/tables.zig index 3992f69d..f4c8b671 100644 --- a/src/state/schema/tables.zig +++ b/src/state/schema/tables.zig @@ -79,6 +79,7 @@ pub fn initCoreTables(db: *sqlite.Db) SchemaError!void { \\ http_proxy_http2_idle_timeout_ms INTEGER, \\ http_proxy_target_port INTEGER, \\ http_proxy_preserve_host INTEGER, + \\ peer_mode TEXT DEFAULT 'off', \\ created_at INTEGER NOT NULL, \\ updated_at INTEGER NOT NULL \\); diff --git a/src/state/store/services_core.zig b/src/state/store/services_core.zig index 3133beb5..3bee558a 100644 --- a/src/state/store/services_core.zig +++ b/src/state/store/services_core.zig @@ -34,7 +34,7 @@ fn createInDb(db: *sqlite.Db, record: ServiceRecord) StoreError!void { var committed = false; errdefer if (!committed) db.exec("ROLLBACK;", .{}, .{}) catch {}; db.exec( - "INSERT INTO services (" ++ service_columns ++ ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);", + "INSERT INTO services (" ++ service_columns ++ ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);", .{}, .{ record.service_name, @@ -53,6 +53,7 @@ fn createInDb(db: *sqlite.Db, record: ServiceRecord) StoreError!void { record.http_proxy_circuit_breaker_threshold, record.http_proxy_circuit_breaker_timeout_ms, record.http_proxy_mirror_service, + record.peer_mode, record.created_at, record.updated_at, }, @@ -140,6 +141,7 @@ pub fn syncConfig( alloc: Allocator, service_name: []const u8, lb_policy: []const u8, + peer_mode: []const u8, routes: []const ServiceHttpRouteInput, ) StoreError!ServiceRecord { var existing = try ensure(alloc, service_name, lb_policy); @@ -154,9 +156,9 @@ pub fn syncConfig( var committed = false; errdefer if (!committed) lease.db.exec("ROLLBACK;", .{}, .{}) catch {}; lease.db.exec( - "UPDATE services SET lb_policy = ?, updated_at = ? WHERE service_name = ?;", + "UPDATE services SET lb_policy = ?, peer_mode = ?, updated_at = ? WHERE service_name = ?;", .{}, - .{ lb_policy, now, service_name }, + .{ lb_policy, peer_mode, now, service_name }, ) catch return StoreError.WriteFailed; try service_routes.replaceInDb(lease.db, service_name, now, routes); try service_routes.syncDerivedFields(lease.db, service_name, now, routes); @@ -313,6 +315,7 @@ test "syncConfig updates proxy policy without changing vip" { alloc, "api", "consistent_hash", + "off", &.{ .{ .route_name = "default", @@ -338,3 +341,31 @@ test "syncConfig updates proxy policy without changing vip" { try std.testing.expectEqual(@as(?bool, false), updated.http_proxy_preserve_host); try std.testing.expectEqualStrings("api-shadow", updated.http_proxy_mirror_service.?); } + +test "syncConfig persists peer_mode round-trip" { + try common.initTestDb(); + defer common.deinitTestDb(); + + const alloc = std.testing.allocator; + + // initial create defaults to "off" (column default). + { + const first = try ensure(alloc, "billing", "consistent_hash"); + defer first.deinit(alloc); + } + + // flip to "require" via syncConfig. + const updated = try syncConfig(alloc, "billing", "consistent_hash", "require", &.{}); + defer updated.deinit(alloc); + try std.testing.expectEqualStrings("require", updated.peer_mode.?); + + // re-read via get to confirm persistence. + const reread = try get(alloc, "billing"); + defer reread.deinit(alloc); + try std.testing.expectEqualStrings("require", reread.peer_mode.?); + + // flip back to "warn". + const downgraded = try syncConfig(alloc, "billing", "consistent_hash", "warn", &.{}); + defer downgraded.deinit(alloc); + try std.testing.expectEqualStrings("warn", downgraded.peer_mode.?); +} diff --git a/src/state/store/services_types.zig b/src/state/store/services_types.zig index d7fe824f..9c8104d9 100644 --- a/src/state/store/services_types.zig +++ b/src/state/store/services_types.zig @@ -23,6 +23,11 @@ pub const ServiceRecord = struct { http_proxy_circuit_breaker_threshold: ?i64 = null, http_proxy_circuit_breaker_timeout_ms: ?i64 = null, http_proxy_mirror_service: ?[]const u8 = null, + /// service-to-service mTLS posture. textual: "off" | "warn" | "require". + /// null means "off" (the column default), which keeps existing record + /// literals working without churn. when non-null the slice is + /// allocator-owned and freed in deinit. + peer_mode: ?[]const u8 = null, created_at: i64, updated_at: i64, @@ -36,11 +41,12 @@ pub const ServiceRecord = struct { if (self.http_proxy_path_prefix) |path_prefix| alloc.free(path_prefix); if (self.http_proxy_rewrite_prefix) |rewrite_prefix| alloc.free(rewrite_prefix); if (self.http_proxy_mirror_service) |mirror_service| alloc.free(mirror_service); + if (self.peer_mode) |peer_mode| alloc.free(peer_mode); } }; pub const service_columns = - "service_name, vip_address, lb_policy, http_proxy_host, http_proxy_path_prefix, http_proxy_rewrite_prefix, http_proxy_retries, http_proxy_connect_timeout_ms, http_proxy_request_timeout_ms, http_proxy_http2_idle_timeout_ms, http_proxy_target_port, http_proxy_preserve_host, http_proxy_retry_on_5xx, http_proxy_circuit_breaker_threshold, http_proxy_circuit_breaker_timeout_ms, http_proxy_mirror_service, created_at, updated_at"; + "service_name, vip_address, lb_policy, http_proxy_host, http_proxy_path_prefix, http_proxy_rewrite_prefix, http_proxy_retries, http_proxy_connect_timeout_ms, http_proxy_request_timeout_ms, http_proxy_http2_idle_timeout_ms, http_proxy_target_port, http_proxy_preserve_host, http_proxy_retry_on_5xx, http_proxy_circuit_breaker_threshold, http_proxy_circuit_breaker_timeout_ms, http_proxy_mirror_service, peer_mode, created_at, updated_at"; pub const ServiceRow = struct { service_name: sqlite.Text, @@ -59,6 +65,7 @@ pub const ServiceRow = struct { http_proxy_circuit_breaker_threshold: ?i64, http_proxy_circuit_breaker_timeout_ms: ?i64, http_proxy_mirror_service: ?sqlite.Text, + peer_mode: ?sqlite.Text, created_at: i64, updated_at: i64, }; @@ -82,6 +89,7 @@ pub fn rowToServiceRecord(row: ServiceRow, http_routes: []const ServiceHttpRoute .http_proxy_circuit_breaker_threshold = row.http_proxy_circuit_breaker_threshold, .http_proxy_circuit_breaker_timeout_ms = row.http_proxy_circuit_breaker_timeout_ms, .http_proxy_mirror_service = if (row.http_proxy_mirror_service) |mirror_service| mirror_service.data else null, + .peer_mode = if (row.peer_mode) |peer_mode| peer_mode.data else null, .created_at = row.created_at, .updated_at = row.updated_at, };