diff --git a/README.md b/README.md index 84109fb..88a1c1e 100644 --- a/README.md +++ b/README.md @@ -151,7 +151,10 @@ trace-specific endpoint is used verbatim. `--otel-service-name` and `OTEL_SERVICE_NAME` set the `service.name` resource attribute, while the build version supplies `service.version`. Standard endpoint, traces-endpoint, header, gzip, insecure, timeout, batch, and sampler environment controls are -documented in [`docs/opentelemetry.md`](docs/opentelemetry.md). +documented in [`docs/opentelemetry.md`](docs/opentelemetry.md). Production +orchestrator, repository, live-batch, block-flush, and rotation work carries +the matching upstream span tree; an offline real-collector receipt exercises +both synchronous and asynchronous archive boundaries. The subscriber's writer-owned hot log retains 256 MiB by default and never evicts rows above the durable archive watermark. Operators can tune the real diff --git a/docs/configuration-parity.md b/docs/configuration-parity.md index e174d99..8c4bab7 100644 --- a/docs/configuration-parity.md +++ b/docs/configuration-parity.md @@ -28,7 +28,7 @@ the same runtime mechanism and an offline receipt observes that behavior. | public/debug bind addresses | closed | `serve --addr` binds the protocol-only public listener, an empty `--debug-addr` binds nothing, and a configured debug listener exclusively serves health/readiness/metrics; `just listener-contract` proves route isolation, public WebSocket upgrade, rejected debug upgrade, disabled-debug refusal, listener readiness, and the historical combined-port compatibility mode through real offline processes | | shutdown/client drain timeouts | closed | canonical signed Go-duration flags, 5s/10s defaults, net/http's non-positive 30s sentinel, the client drain's already-expired non-positive sentinel (immediate teardown with a best-effort close frame), listener-admission gating, concurrent public/debug/pipeline shutdown, close code 1001 for positive budgets, cooperative early exit, and one shared silent-client deadline are covered by unit and real ReleaseSafe process receipts. Per-client close hooks now fan out concurrently with borrowed-handler lifetime protection; a real-socket transport regression saturates the first peer's send buffer, proves a later peer receives 1001 before the deadline, and proves the blocked writer is interrupted at the shared deadline. | | log level/format | closed | `--log-level` retains debug calls in ReleaseSafe and applies the upstream case-insensitive aliases plus info default to every `std.log` producer; `--log-format` defaults to newline-delimited JSON and selects canonical text fields. `just logging-contract` proves default JSON, warn filtering, text/debug output, persistent root-flag placement, and fail-closed invalid values through real offline processes. | -| OpenTelemetry | open | provider bootstrap is real and empirically covered: endpoint activation/precedence, HTTP/protobuf path rules, headers, gzip, insecure mode, exporter timeout, batch queue/delay/size, the six SDK samplers, process plus service resource attributes, global provider installation, and bounded shutdown export all run through pinned `otel-zig`. Remaining blockers are upstream-equivalent span boundaries, TraceContext+Baggage propagation, exporter retry, BSP export timeout, and custom CA/mTLS credentials. See `opentelemetry.md`. | +| OpenTelemetry | open | provider bootstrap is real and empirically covered: endpoint activation/precedence, HTTP/protobuf path rules, headers, gzip, insecure mode, exporter timeout, batch queue/delay/size, the six SDK samplers, process plus service resource attributes, global provider installation, and bounded shutdown export all run through pinned `otel-zig`. All orchestrator, backfill/resync, live-batch, block-flush, and rotation spans now match upstream names, scopes, parentage, attributes, and success/error finalization; a real offline OTLP collector exercises the production sync/async archive boundaries. Remaining blockers are the `xrpcapi/getBlock` result span, TraceContext+Baggage propagation, exporter retry, BSP export timeout, and custom CA/mTLS credentials. See `opentelemetry.md`. | | `JETSTREAM_*` environment sources and unknown-variable rejection | closed | the exact pinned flag-source map applies root, serve, and `inspect-all` values with CLI-over-environment precedence; overridden invalid values are never parsed. Unknown names are sorted/deduplicated and rejected before command execution while upstream's Kubernetes `JETSTREAM_APP_*` and simulator `JETSTREAM_SIM_*` namespaces are exempt. `just environment-contract` proves real ReleaseSafe root/inspect/serve processes, representative string behavior, every typed source, timestamp-import authentication, and precedence offline. | | serve/version/inspect command surface | closed | `serve` is accepted explicitly while retaining Stream's historical flag-only invocation. `version` emits the upstream build-information shape. `inspect-segment` matches the pinned renderer byte-for-byte on an upstream-produced sealed fixture while also walking active files, surfacing partial tails, and labelling readable checksum corruption. `inspect-all` folds the real steady and bootstrap trees with upstream's missing-root, racing-tail, active-skip, aggregation, sorting, truncation, and text-rendering semantics; a copied upstream golden report is byte-exact. Listener and root logging controls remain tracked by their separate rows. | diff --git a/docs/opentelemetry.md b/docs/opentelemetry.md index 70a1861..fc368a0 100644 --- a/docs/opentelemetry.md +++ b/docs/opentelemetry.md @@ -31,6 +31,24 @@ not select JSON there and do not select JSON in Stream. The executable accepts `--otel-service-name`; `OTEL_SERVICE_NAME` supplies its default and `jetstream` is the fallback. CLI values take precedence. +## Production span boundaries + +Stream now carries one explicit trace context across the same asynchronous and +durability boundaries as pinned upstream. The implemented span names are: + +| Scope | Spans | +|---|---| +| `jetstream/ingest/orchestrator` | `Run`, `runBootstrap`, `finishBootstrap`, `runMerge`, merge-runner `run`, `processSourceSegment`, `runDiscovery`, `runSteadyState`, `runDeleteCompaction`, `RunImport` | +| `jetstream/ingest/backfill` | `Run`, `handleRepo` (including retry and live resync, with `did`) | +| `jetstream/ingest/live` | `processBatch` | +| `jetstream/ingest` | `commitAsyncFlush`, `rotateIfFull`, `flushAndRotateLocked`, `rotateLocked` (with `active_idx`) | + +Spans end at the real operation boundary, record returned errors, and mark +successful operations `OK`. Repository downloads remain children of the +backfill run; `handleRepo` begins only after a CAR has been accepted, matching +upstream's materialization boundary. Writer spans occur per block or rotation, +never per event. + ## Offline receipt The `internal/observability.zig` contract starts a real loopback HTTP @@ -41,13 +59,19 @@ endpoint path, configured header, instrumentation scope, span name, tests cover no-op behavior and environment precedence/fallbacks. This receipt requires no external network. +`internal/tracing_integration_test.zig` then drives the production archive in +both asynchronous and synchronous modes. A real OTLP collector requires the +exported parent `processBatch` plus `commitAsyncFlush`, `rotateIfFull`, +`rotateLocked`, and `flushAndRotateLocked` children and both canonical scopes. +This is an offline real-I/O receipt, not a recording exporter or span mock. + ## Remaining parity gate The provider is not the entire tracing contract. A full-network experiment remains blocked until Stream also has: -1. spans at the same per-repository, per-batch, per-seal, and orchestrator - boundaries as pinned upstream, with matching success/error status; +1. the `jetstream/xrpcapi` `getBlock` span, including upstream's `result` + attribute and error status; 2. global W3C TraceContext plus Baggage propagation; 3. the upstream export retry policy and `OTEL_BSP_EXPORT_TIMEOUT` behavior; and 4. custom CA and client certificate/key handling for the corresponding OTLP diff --git a/src/internal/archive.zig b/src/internal/archive.zig index 9c6f928..e7ecc09 100644 --- a/src/internal/archive.zig +++ b/src/internal/archive.zig @@ -17,6 +17,7 @@ const std = @import("std"); const segment = @import("segment.zig"); const writer_mod = @import("segment_writer.zig"); const metrics = @import("metrics.zig"); +const observability = @import("observability.zig"); const manifest_mod = @import("manifest.zig"); const segment_io = @import("segment_io.zig"); @@ -418,15 +419,33 @@ pub const Archive = struct { /// append one event; assigns and returns its local seq. blocks flush to /// disk at the block threshold; rotation seals at the byte budget. pub fn append(self: *Archive, ev: segment.Event, now_us: i64) !u64 { + return self.appendTraced(ev, now_us, observability.root_context); + } + + pub fn appendTraced( + self: *Archive, + ev: segment.Event, + now_us: i64, + trace_context: observability.Context, + ) !u64 { var rows = [_]segment.Event{ev}; - return (try self.appendBatch(&rows, now_us)).?; + return (try self.appendBatchTraced(&rows, now_us, trace_context)).?; } /// Append every row derived from one upstream event under one writer /// lock. This is the transaction boundary relied on by frame seq /// assignment (`first_seq + op_index`) and verifier checkpoints. pub fn appendBatch(self: *Archive, events: []segment.Event, now_us: i64) !?u64 { - return self.appendBatchBeforeDurable(events, now_us, null, null); + return self.appendBatchTraced(events, now_us, observability.root_context); + } + + pub fn appendBatchTraced( + self: *Archive, + events: []segment.Event, + now_us: i64, + trace_context: observability.Context, + ) !?u64 { + return self.appendBatchBeforeDurableTraced(events, now_us, null, null, trace_context); } /// Append rows and run `before_durable(first,last)` under the writer lock @@ -439,11 +458,28 @@ pub const Archive = struct { now_us: i64, hook_ctx: ?*anyopaque, before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void, + ) !?u64 { + return self.appendBatchBeforeDurableTraced( + events, + now_us, + hook_ctx, + before_durable, + observability.root_context, + ); + } + + pub fn appendBatchBeforeDurableTraced( + self: *Archive, + events: []segment.Event, + now_us: i64, + hook_ctx: ?*anyopaque, + before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void, + trace_context: observability.Context, ) !?u64 { if (events.len == 0) return null; if (self.async_flush != null) - return @as(?u64, try self.appendBatchAsync(events, now_us, hook_ctx, before_durable)); - return @as(?u64, try self.appendBatchSync(events, now_us, hook_ctx, before_durable)); + return @as(?u64, try self.appendBatchAsync(events, now_us, hook_ctx, before_durable, trace_context)); + return @as(?u64, try self.appendBatchSync(events, now_us, hook_ctx, before_durable, trace_context)); } fn appendBatchSync( @@ -452,7 +488,11 @@ pub const Archive = struct { now_us: i64, hook_ctx: ?*anyopaque, before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void, + trace_context: observability.Context, ) !u64 { + var flush_trace: ?observability.Span = null; + defer if (flush_trace) |*trace| trace.deinit(); + errdefer |err| if (flush_trace) |*trace| trace.fail(err); self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); errdefer self.poisoned = true; @@ -463,13 +503,30 @@ pub const Archive = struct { } if (before_durable) |hook| try hook(hook_ctx.?, first_seq, self.next_seq - 1); if (self.writer.pending.items.len > 0 and now_us - self.last_sync_us >= self.flush_interval_us) { + flush_trace = observability.Span.start( + self.allocator, + "ingest", + "flushAndRotateLocked", + trace_context, + &.{}, + ); try self.writer.flush(); // partial block: durability over block size } if (self.writer.bytes().len > self.file_written) { + if (flush_trace == null) + flush_trace = observability.Span.start( + self.allocator, + "ingest", + "flushAndRotateLocked", + trace_context, + &.{}, + ); try self.syncToDisk(null); self.last_sync_us = now_us; - if (self.writer.bytes().len >= self.max_segment_bytes) try self.rotateLocked(); + if (self.writer.bytes().len >= self.max_segment_bytes) + try self.rotateLockedTraced(flush_trace.?.context()); } + if (flush_trace) |*trace| trace.succeed(); return first_seq; } @@ -479,6 +536,7 @@ pub const Archive = struct { now_us: i64, hook_ctx: ?*anyopaque, before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void, + trace_context: observability.Context, ) !u64 { const pipeline = &self.async_flush.?; pipeline.admission_mu.lockUncancelable(self.io); @@ -519,34 +577,46 @@ pub const Archive = struct { pipeline.admission_mu.unlock(self.io); admission_locked = false; - try self.finishAsyncJobs(jobs); + try self.finishAsyncJobs(jobs, trace_context); if (jobs.len > 0) { self.mu.lockUncancelable(self.io); self.last_sync_us = now_us; self.mu.unlock(self.io); } - try self.rotateIfFullAsync(); + try self.rotateIfFullAsync(trace_context); return first_seq; } - fn finishAsyncJobs(self: *Archive, jobs: []AsyncFlushJob) !void { + fn finishAsyncJobs(self: *Archive, jobs: []AsyncFlushJob, trace_context: observability.Context) !void { const pipeline = &self.async_flush.?; var first_error: ?anyerror = null; for (jobs) |*job| { job.compressed.waitUncancelable(self.io); pipeline.waitTurn(job.id); + var trace = observability.Span.start( + self.allocator, + "ingest", + "commitAsyncFlush", + trace_context, + &.{}, + ); + defer trace.deinit(); self.mu.lockUncancelable(self.io); if (job.failure) |err| { + trace.fail(err); self.poisoned = true; if (first_error == null) first_error = err; } else if (self.poisoned) { + trace.fail(error.ArchivePoisoned); if (first_error == null) first_error = error.ArchivePoisoned; } else { self.writer.commitPrepared(job.frame.?) catch |err| { + trace.fail(err); self.poisoned = true; if (first_error == null) first_error = err; }; if (!self.poisoned) self.syncToDisk(job.prepared.max_seq) catch |err| { + trace.fail(err); self.poisoned = true; if (first_error == null) first_error = err; }; @@ -554,17 +624,23 @@ pub const Archive = struct { self.mu.unlock(self.io); if (job.frame) |frame| self.allocator.free(frame); pipeline.finishTurn(); + trace.succeed(); } if (first_error) |err| return err; } - fn rotateIfFullAsync(self: *Archive) !void { + fn rotateIfFullAsync(self: *Archive, trace_context: observability.Context) !void { const pipeline = &self.async_flush.?; self.mu.lockUncancelable(self.io); const full = self.writer.bytes().len >= self.max_segment_bytes; self.mu.unlock(self.io); if (!full) return; + var trace = observability.Span.start(self.allocator, "ingest", "rotateIfFull", trace_context, &.{}); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); + pipeline.admission_mu.lockUncancelable(self.io); defer pipeline.admission_mu.unlock(self.io); pipeline.drain(); @@ -572,7 +648,7 @@ pub const Archive = struct { defer self.mu.unlock(self.io); errdefer self.poisoned = true; if (self.poisoned) return error.ArchivePoisoned; - if (self.writer.bytes().len >= self.max_segment_bytes) try self.rotateLocked(); + if (self.writer.bytes().len >= self.max_segment_bytes) try self.rotateLockedTraced(trace.context()); } fn appendStampedLocked(self: *Archive, event: *segment.Event) !u64 { @@ -647,14 +723,22 @@ pub const Archive = struct { /// compress + write any pending block, fsync, advance the watermark. pub fn flushBlock(self: *Archive) !void { - if (self.async_flush != null) return self.flushBlockAsync(); + return self.flushBlockTraced(observability.root_context); + } + + pub fn flushBlockTraced(self: *Archive, trace_context: observability.Context) !void { + if (self.async_flush != null) return self.flushBlockAsync(trace_context); + var trace = observability.Span.start(self.allocator, "ingest", "flushAndRotateLocked", trace_context, &.{}); + defer trace.deinit(); + defer trace.succeed(); + errdefer |err| trace.fail(err); self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); errdefer self.poisoned = true; try self.flushBlockLocked(); } - fn flushBlockAsync(self: *Archive) !void { + fn flushBlockAsync(self: *Archive, trace_context: observability.Context) !void { const pipeline = &self.async_flush.?; pipeline.admission_mu.lockUncancelable(self.io); var admission_locked = true; @@ -675,7 +759,7 @@ pub const Archive = struct { pipeline.submit(&job); pipeline.admission_mu.unlock(self.io); admission_locked = false; - try self.finishAsyncJobs((&job)[0..1]); + try self.finishAsyncJobs((&job)[0..1], trace_context); } fn flushBlockLocked(self: *Archive) !void { @@ -716,6 +800,18 @@ pub const Archive = struct { try self.rotateLocked(); } + pub fn rotateTraced(self: *Archive, trace_context: observability.Context) !void { + if (self.async_flush) |*pipeline| { + pipeline.admission_mu.lockUncancelable(self.io); + defer pipeline.admission_mu.unlock(self.io); + pipeline.drain(); + } + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + errdefer self.poisoned = true; + try self.rotateLockedTraced(trace_context); + } + fn rotateLocked(self: *Archive) !void { const seal_start = Io.Timestamp.now(self.io, .awake); try self.flushBlockLocked(); @@ -738,6 +834,15 @@ pub const Archive = struct { try self.openNextSegment(); } + fn rotateLockedTraced(self: *Archive, trace_context: observability.Context) !void { + var trace = observability.Span.start(self.allocator, "ingest", "rotateLocked", trace_context, &.{}); + defer trace.deinit(); + defer trace.succeed(); + errdefer |err| trace.fail(err); + try self.rotateLocked(); + trace.setAttribute(.{ .key = "active_idx", .value = .{ .int = @intCast(self.seg_index) } }); + } + /// terminal cutover (bootstrap-live source at merge): seal the active /// segment so the merge drain can read it; an empty active (header /// only, no blocks) is deleted instead. the archive is unusable after. diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig index 6357637..a104a6f 100644 --- a/src/internal/bootstrap/engine.zig +++ b/src/internal/bootstrap/engine.zig @@ -21,6 +21,7 @@ const fetched_car = @import("fetched_car.zig"); const prepared_fetch = @import("prepared_fetch.zig"); const metrics_mod = @import("../metrics.zig"); const meta_store = @import("../meta_store.zig"); +const observability = @import("../observability.zig"); const repo_store = @import("repo_store.zig"); const repos = @import("repos.zig"); const segment = @import("../segment.zig"); @@ -231,8 +232,15 @@ const Work = struct { finished: usize = 0, fatal_error: ?anyerror = null, prepare_mu: Io.Mutex = .init, - - fn init(allocator: Allocator, io: Io, config: Config, archive: *archive_mod.Archive) !Work { + trace_context: observability.Context, + + fn init( + allocator: Allocator, + io: Io, + config: Config, + archive: *archive_mod.Archive, + trace_context: observability.Context, + ) !Work { if (config.max_inflight_bytes == 0) return error.InvalidBackfillLimits; const worker_count = effectiveWorkerCount(config.workers); const queue_capacity = worker_count * 2; @@ -244,6 +252,7 @@ const Work = struct { .queue = .{ .buffer = try allocator.alloc(Job, queue_capacity) }, .budget = byte_budget.Budget.init(io, config.max_inflight_bytes), .worker_count = worker_count, + .trace_context = trace_context, }; if (config.metrics) |m| { work.budget.used_gauge = &m.backfill_inflight_bytes; @@ -337,6 +346,7 @@ const Work = struct { self.archive, &self.prepare_mu, self.config.metrics, + self.trace_context, ) catch |err| { if (self.config.metrics) |m| _ = m.backfill_workers_active.fetchSub(1, .monotonic); if (isFatalDownloadError(err)) { @@ -512,6 +522,21 @@ pub fn run( archive: *archive_mod.Archive, store: *repo_store.Store, ) !Stats { + return runTraced(allocator, io, config, archive, store, observability.root_context); +} + +pub fn runTraced( + allocator: Allocator, + io: Io, + config: Config, + archive: *archive_mod.Archive, + store: *repo_store.Store, + trace_context: observability.Context, +) !Stats { + var trace = observability.Span.start(allocator, "ingest/backfill", "Run", trace_context, &.{}); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); if (config.selected_repos.len > 0 and config.max_repos > 0) return error.ConflictingBackfillSelection; if (config.selected_repos.len > 0 and config.plc_url == null) return error.SelectedBackfillRequiresIdentityResolver; var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; @@ -526,7 +551,7 @@ pub fn run( // parallel whole-network engine. var work_config = config; if (config.selected_repos.len > 0) work_config.workers = 1; - var work = try Work.init(allocator, io, work_config, archive); + var work = try Work.init(allocator, io, work_config, archive, trace.context()); defer work.deinit(); try work.start(); defer work.shutdown(); @@ -994,6 +1019,7 @@ fn downloadOne( archive: *archive_mod.Archive, prepare_mu: *Io.Mutex, instrumentation: ?*metrics_mod.Stats, + trace_context: observability.Context, ) !Result { var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; const scratch_dir = try std.fmt.bufPrint(&scratch_buf, "{s}/backfill/repo-scratch", .{data_dir}); @@ -1038,6 +1064,19 @@ fn downloadOne( }, .ok => |*prepared| { defer prepared.deinit(io); + const attributes = [_]observability.Attribute{ + .{ .key = "did", .value = .{ .string = did } }, + }; + var trace = observability.Span.start( + result_allocator, + "ingest/backfill", + "handleRepo", + trace_context, + &attributes, + ); + defer trace.deinit(); + defer trace.succeed(); + errdefer |err| trace.fail(err); const handle_started = Io.Timestamp.now(io, .awake); var local: Stats = .{}; const now_us: i64 = Io.Timestamp.now(io, .real).toMicroseconds(); @@ -1047,10 +1086,15 @@ fn downloadOne( batch: [1024]segment.Event = undefined, batch_len: usize = 0, rows: u64 = 0, + trace_context: observability.Context, fn flush(self: *@This()) anyerror!void { if (self.batch_len == 0) return; - _ = try self.archive.appendBatch(self.batch[0..self.batch_len], self.now_us); + _ = try self.archive.appendBatchTraced( + self.batch[0..self.batch_len], + self.now_us, + self.trace_context, + ); self.batch_len = 0; } @@ -1061,9 +1105,13 @@ fn downloadOne( if (self.batch_len == self.batch.len) try self.flush(); } }; - var sink: Sink = .{ .archive = archive, .now_us = now_us }; + var sink: Sink = .{ + .archive = archive, + .now_us = now_us, + .trace_context = trace.context(), + }; const head_rev = repos.emitPrepared(alloc, &prepared.repo, did, now_us, .create, &local.drops, &sink, Sink.emit) catch |err| switch (err) { - error.OutOfMemory, error.ArchiveAppendFailed => return err, + error.OutOfMemory, error.ArchiveAppendFailed => return @errorCast(err), error.MissingBlock, error.IncompleteCar => { if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, .car, @errorName(err)); try sleepBackoff(io, transient_attempt, 30); @@ -1073,7 +1121,7 @@ fn downloadOne( else => return failureResult(result_allocator, did, &final_host, attempts, started_us, .car, @errorName(err)), }; sink.flush() catch |err| switch (err) { - error.OutOfMemory, error.ArchiveAppendFailed => return err, + error.OutOfMemory, error.ArchiveAppendFailed => return @errorCast(err), else => return error.ArchiveAppendFailed, }; if (instrumentation) |m| m.observeBackfillHandleRepo(@intCast(@max( @@ -1277,6 +1325,7 @@ test "bootstrap redirects preserve independent transient and rate-limit retry bu &archive, &convert_mu, &instrumentation, + observability.root_context, ); defer freeResult(testing.allocator, result); @@ -1362,6 +1411,7 @@ test "successful real CAR handling records only the handler boundary" { &archive, &convert_mu, &instrumentation, + observability.root_context, ); defer freeResult(testing.allocator, result); diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index 4b15efe..3b28f04 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -27,6 +27,7 @@ const ingest = @import("../ingest.zig"); const merge = @import("merge.zig"); const metrics = @import("../metrics.zig"); const meta_store = @import("../meta_store.zig"); +const observability = @import("../observability.zig"); const phase_mod = @import("phase.zig"); const pipeline_mod = @import("../pipeline.zig"); const repair_mod = @import("../repair.zig"); @@ -80,6 +81,20 @@ pub fn run( cursor_store: *cursor_mod.Store, verifier: ?*verify.Verifier, stats: *metrics.Stats, +) !void { + return runTraced(allocator, io, opts, dst, meta, cursor_store, verifier, stats, observability.root_context); +} + +pub fn runTraced( + allocator: Allocator, + io: Io, + opts: Options, + dst: *archive_mod.Archive, + meta: *meta_store.Store, + cursor_store: *cursor_mod.Store, + verifier: ?*verify.Verifier, + stats: *metrics.Stats, + trace_context: observability.Context, ) !void { var phase_store = phase_mod.Store.init(meta); defer phase_store.deinit(io); @@ -111,7 +126,28 @@ pub fn run( defer repo_st.deinit(); if (phase == .bootstrap) { - try runBootstrap(allocator, io, opts, live_data_dir, dst, cursor_store, verifier, stats, &repo_st); + var bootstrap_trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "runBootstrap", + trace_context, + &.{}, + ); + defer bootstrap_trace.deinit(); + defer bootstrap_trace.succeed(); + errdefer |err| bootstrap_trace.fail(err); + try runBootstrap( + allocator, + io, + opts, + live_data_dir, + dst, + cursor_store, + verifier, + stats, + &repo_st, + bootstrap_trace.context(), + ); crashpoint.hit(.after_backfill_before_phase_merging); const write_started = Io.Timestamp.now(io, .awake); try phase_store.write(io, .merging); @@ -119,8 +155,19 @@ pub fn run( stats.observeOrchestratorTransition(.bootstrap, .merging); stats.setOrchestratorPhase(.merging); phase = .merging; + bootstrap_trace.succeed(); } + var merge_trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "runMerge", + trace_context, + &.{}, + ); + defer merge_trace.deinit(); + defer merge_trace.succeed(); + errdefer |err| merge_trace.fail(err); const merge_started = Io.Timestamp.now(io, .awake); var merge_observed = false; defer if (!merge_observed) stats.observeOrchestratorState(.merge, elapsedUs(io, merge_started)); @@ -140,7 +187,7 @@ pub fn run( var bf_dir = try Io.Dir.cwd().openDir(io, bf_path, .{}); defer bf_dir.close(io); - const m = try merge.run(allocator, io, live_data_dir, dst, &repo_st, stats); + const m = try merge.runTraced(allocator, io, live_data_dir, dst, &repo_st, stats, merge_trace.context()); log.info("merge: {d} kept, {d} dropped across {d} sources", .{ m.kept, m.dropped, m.sources }); try dst.flushBlock(); crashpoint.hit(.after_merge_drain_before_pending); @@ -149,7 +196,7 @@ pub fn run( // whose bootstrap download was crash-interrupted resync NOW, // above the merged live tail, so the first served view has their // current state before serving ungates - try runPendingPass(allocator, io, opts.data_dir, opts.relay_http, dst, &repo_st); + try runPendingPass(allocator, io, opts.data_dir, opts.relay_http, dst, &repo_st, merge_trace.context()); crashpoint.hit(.after_pending_pass); } @@ -159,6 +206,16 @@ pub fn run( // nothing is serving yet. idempotent, so the restart-after-cleanup // path re-runs it too. try dst.rotate(); + var compaction_trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "runDeleteCompaction", + merge_trace.context(), + &.{}, + ); + defer compaction_trace.deinit(); + defer compaction_trace.succeed(); + errdefer |err| compaction_trace.fail(err); if (opts.compaction_enabled) { const compaction_started = Io.Timestamp.now(io, .awake); var seg_path_buf: [256]u8 = undefined; @@ -170,13 +227,14 @@ pub fn run( .measure_pass = false, }) catch |err| { stats.observeCompactionPass(.error_result, elapsedUs(io, compaction_started)); - return err; + return @errorCast(err); }; stats.observeCompactionPass(.ok, elapsedUs(io, compaction_started)); log.info("merge-tail compaction: watermark {d}, {d} chunks, {d}/{d} segments rewritten, {d} rows dropped", .{ cs.target_watermark, cs.chunks, cs.segments_rewritten, cs.segments_examined, cs.rows_dropped, }); } + compaction_trace.succeed(); // Merge-tail rewriting is deliberately manifest-oblivious. Before // serving can ungate, every resident header/index must match disk. const reconciled = try dst.reconcileManifest(); @@ -194,7 +252,7 @@ pub fn run( // outage here fails the merge (upstream couples them the same // way) — re-entry is idempotent. if (!opts.skip_merge_discovery) { - try runDiscovery(allocator, io, opts.relay_http, &repo_st, stats); + try runDiscovery(allocator, io, opts.relay_http, &repo_st, stats, merge_trace.context()); } else { log.info("merge discovery: skipped by configuration", .{}); } @@ -208,7 +266,7 @@ pub fn run( // handle is a posix fd) meta_dir.deleteTree(io, "backfill") catch |err| { log.err("merge cleanup: remove backfill tree: {s}", .{@errorName(err)}); - return err; + return @errorCast(err); }; crashpoint.hit(.after_cleanup_remove_before_sync); } @@ -218,7 +276,7 @@ pub fn run( const dir_file: Io.File = .{ .handle = meta_dir.handle, .flags = .{ .nonblocking = false } }; dir_file.sync(io) catch |err| { log.err("merge cleanup: sync data directory: {s}", .{@errorName(err)}); - return err; + return @errorCast(err); }; // Cleanup's dirent must be durable before these synced cursor deletes. // Otherwise a power cut can resurrect the source tree with cursor zero @@ -230,7 +288,7 @@ pub fn run( cleanup.delete("bootstrap/last_listrepos_cursor"); cleanup.commitDurable() catch |err| { log.err("merge cleanup: delete durable cursors: {s}", .{@errorName(err)}); - return err; + return @errorCast(err); }; } crashpoint.hit(.after_merge_cleanup_complete); @@ -252,6 +310,7 @@ pub fn run( } stats.observeOrchestratorState(.merge, elapsedUs(io, merge_started)); merge_observed = true; + merge_trace.succeed(); const write_started = Io.Timestamp.now(io, .awake); try phase_store.write(io, .steady_state); stats.observeOrchestratorState(.write_phase_steady, elapsedUs(io, write_started)); @@ -270,6 +329,7 @@ fn runPendingPass( relay_http: []const u8, dst: *archive_mod.Archive, repo_st: *repo_store_mod.Store, + trace_context: observability.Context, ) !void { const dids = try repo_st.collectByStatus(allocator, .pending); defer repo_st.freeDids(allocator, dids); @@ -282,7 +342,7 @@ fn runPendingPass( const scratch_dir = try std.fmt.bufPrint(&scratch_buf, "{s}/backfill/repo-scratch", .{data_dir}); for (dids) |did| { var details: resync.AttemptDetails = .{}; - _ = resync.resyncRepo(allocator, io, scratch_dir, relay_http, did, dst, repo_st, &drops, &details) catch |err| { + _ = resync.resyncRepo(allocator, io, scratch_dir, relay_http, did, dst, repo_st, &drops, &details, trace_context) catch |err| { if (details.local_failure) return err; switch (err) { error.OutOfMemory => return err, @@ -329,7 +389,18 @@ fn runDiscovery( relay_http: []const u8, repo_st: *repo_store_mod.Store, stats: *metrics.Stats, + trace_context: observability.Context, ) !void { + var trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "runDiscovery", + trace_context, + &.{}, + ); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); const start = (try engine.loadBootstrapLastListReposCursor(repo_st.meta, allocator)) orelse { log.info("merge discovery: no bootstrap cursor, skipping", .{}); return; @@ -382,6 +453,7 @@ fn runBootstrap( verifier: ?*verify.Verifier, stats: *metrics.Stats, repo_st: *repo_store_mod.Store, + trace_context: observability.Context, ) !void { var live_archive = try archive_mod.Archive.init(allocator, io, live_data_dir); live_archive.stats = stats; @@ -421,6 +493,7 @@ fn runBootstrap( .emit_repair = ingest.Consumer.emitRepair, .emit_cursor = ingest.Consumer.emitCursor, .cursor_watermark = cursor_store.load(io) orelse 0, + .trace_context = trace_context, }; var repair_scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; const repair_scratch = try std.fmt.bufPrint(&repair_scratch_buf, "{s}/repair-scratch", .{live_data_dir}); @@ -462,7 +535,7 @@ fn runBootstrap( log.info("backfill: async block compression off", .{}); var async_flush_enabled = opts.async_flush_workers > 0; defer if (async_flush_enabled) dst.disableAsyncFlush(); - const bf = engine.run(allocator, io, .{ + const bf = engine.runTraced(allocator, io, .{ .relay_http = opts.relay_http, .plc_url = opts.plc_url, .data_dir = opts.data_dir, @@ -472,7 +545,7 @@ fn runBootstrap( .batch_size = opts.batch_size, .max_inflight_bytes = opts.max_inflight_bytes, .metrics = stats, - }, dst, repo_st) catch |err| { + }, dst, repo_st, trace_context) catch |err| { capture.requestStop(); _ = capture_future.cancel(io); return err; @@ -490,6 +563,16 @@ fn runBootstrap( // cancel unblocks an idle read), then seal its archive so the merge // reads a fully-sealed source tree (the merge's seal guard covers a // crash landing between these two steps) + var finish_trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "finishBootstrap", + trace_context, + &.{}, + ); + defer finish_trace.deinit(); + defer finish_trace.succeed(); + errdefer |err| finish_trace.fail(err); const drain_started = Io.Timestamp.now(io, .awake); capture.requestStop(); _ = capture_future.cancel(io); @@ -507,7 +590,7 @@ fn runBootstrap( crashpoint.hit(.after_live_close_before_seal); live_archive.sealAndClose() catch |err| { log.warn("bootstrap: live archive seal failed: {s}", .{@errorName(err)}); - return err; + return @errorCast(err); }; stats.observeOrchestratorState(.seal_bootstrap, elapsedUs(io, seal_started)); live_sealed = true; diff --git a/src/internal/bootstrap/merge.zig b/src/internal/bootstrap/merge.zig index 0c37be2..143cb6e 100644 --- a/src/internal/bootstrap/merge.zig +++ b/src/internal/bootstrap/merge.zig @@ -18,6 +18,7 @@ const std = @import("std"); const archive_mod = @import("../archive.zig"); const crashpoint = @import("../crashpoint.zig"); const metrics = @import("../metrics.zig"); +const observability = @import("../observability.zig"); const repo_store = @import("repo_store.zig"); const segment = @import("../segment.zig"); @@ -82,6 +83,22 @@ pub fn run( store: *repo_store.Store, metric_stats: *metrics.Stats, ) !Stats { + return runTraced(allocator, io, live_data_dir, dst, store, metric_stats, observability.root_context); +} + +pub fn runTraced( + allocator: Allocator, + io: Io, + live_data_dir: []const u8, + dst: *archive_mod.Archive, + store: *repo_store.Store, + metric_stats: *metrics.Stats, + trace_context: observability.Context, +) !Stats { + var trace = observability.Span.start(allocator, "ingest/orchestrator", "run", trace_context, &.{}); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); // seal guard: a crash between live-consumer teardown and seal leaves // an active trailing source. Archive recovery seals torn actives; // sealAndClose then deletes the fresh empty one it opened. @@ -116,9 +133,30 @@ pub fn run( var source_arena = std.heap.ArenaAllocator.init(allocator); defer source_arena.deinit(); - const revs = try drainOne(source_arena.allocator(), io, src_dir, idx, dst, &cache, &stats, metric_stats); + var source_trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "processSourceSegment", + trace.context(), + &.{}, + ); + defer source_trace.deinit(); + defer source_trace.succeed(); + errdefer |err| source_trace.fail(err); + const revs = try drainOne( + source_arena.allocator(), + io, + src_dir, + idx, + dst, + &cache, + &stats, + metric_stats, + source_trace.context(), + ); // durability ordering: dst rows fsynced before the cursor advances - try dst.flushBlock(); + try dst.flushBlockTraced(source_trace.context()); + source_trace.succeed(); crashpoint.hit(.after_merge_dst_flush_before_source_commit); const updates = try source_arena.allocator().alloc(repo_store.RevUpdate, revs.count()); var update_it = revs.iterator(); @@ -142,6 +180,7 @@ fn drainOne( cache: *RepoStatusCache, stats: *Stats, metric_stats: *metrics.Stats, + trace_context: observability.Context, ) !std.StringHashMapUnmanaged([]const u8) { var name_buf: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buf, idx); @@ -164,7 +203,7 @@ fn drainOne( } var stamped = ev; stamped.witnessed_at = now_us; // upstream §3.4 re-stamp - _ = try dst.append(stamped, now_us); + _ = try dst.appendTraced(stamped, now_us, trace_context); stats.kept += 1; _ = metric_stats.orchestrator_merge_events_kept_total.fetchAdd(1, .monotonic); switch (ev.kind) { diff --git a/src/internal/bootstrap/resync.zig b/src/internal/bootstrap/resync.zig index 4ec4b22..f2f5095 100644 --- a/src/internal/bootstrap/resync.zig +++ b/src/internal/bootstrap/resync.zig @@ -15,6 +15,7 @@ const std = @import("std"); const zat = @import("zat"); const archive_mod = @import("../archive.zig"); +const observability = @import("../observability.zig"); const fetched_car = @import("fetched_car.zig"); const prepared_fetch = @import("prepared_fetch.zig"); const repo_store = @import("repo_store.zig"); @@ -55,6 +56,7 @@ pub fn resyncRepo( store: *repo_store.Store, drops: *repos.DropCounts, details: *AttemptDetails, + trace_context: observability.Context, ) !Stats { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); @@ -103,6 +105,14 @@ pub fn resyncRepo( defer prepared_fetch_result.deinit(io); details.local_failure = true; + const attributes = [_]observability.Attribute{ + .{ .key = "did", .value = .{ .string = did } }, + }; + var trace = observability.Span.start(allocator, "ingest/backfill", "handleRepo", trace_context, &attributes); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); + const now_us: i64 = Io.Timestamp.now(io, .real).toMicroseconds(); const head_rev = prepared_fetch_result.repo.rev; @@ -116,10 +126,11 @@ pub fn resyncRepo( now_us: i64, batch: std.ArrayList(segment.Event) = .empty, row_count: u64 = 0, + trace_context: observability.Context, fn flush(self: *@This()) anyerror!void { if (self.batch.items.len == 0) return; - _ = try self.archive.appendBatch(self.batch.items, self.now_us); + _ = try self.archive.appendBatchTraced(self.batch.items, self.now_us, self.trace_context); self.batch.clearRetainingCapacity(); } @@ -129,12 +140,17 @@ pub fn resyncRepo( if (self.batch.items.len == batch_size) try self.flush(); } }; - var sink: Sink = .{ .archive = archive, .allocator = alloc, .now_us = now_us }; + var sink: Sink = .{ + .archive = archive, + .allocator = alloc, + .now_us = now_us, + .trace_context = trace.context(), + }; defer sink.batch.deinit(alloc); try sink.emit(sync_ev); _ = try repos.emitPrepared(alloc, &prepared_fetch_result.repo, did, now_us, .create_resync, drops, &sink, Sink.emit); try sink.flush(); - try archive.flushBlock(); + try archive.flushBlockTraced(trace.context()); // durability ordering: every replacement batch is durable before the // completion row that describes it. diff --git a/src/internal/bootstrap/retry.zig b/src/internal/bootstrap/retry.zig index bdb4e7d..facd5b4 100644 --- a/src/internal/bootstrap/retry.zig +++ b/src/internal/bootstrap/retry.zig @@ -15,6 +15,7 @@ const std = @import("std"); const archive_mod = @import("../archive.zig"); const metrics = @import("../metrics.zig"); const meta_store = @import("../meta_store.zig"); +const observability = @import("../observability.zig"); const repo_store = @import("repo_store.zig"); const repos = @import("repos.zig"); const resync = @import("resync.zig"); @@ -71,6 +72,7 @@ pub const Retry = struct { stop_flag: std.atomic.Value(bool) = .init(false), stop_event: Io.Event = .unset, stats: ?*metrics.Stats = null, + trace_context: observability.Context = observability.root_context, emit_failure_logs: bool = true, jitter_ctx: ?*anyopaque = null, jitter: ?*const fn (?*anyopaque, u64) u64 = null, @@ -277,7 +279,7 @@ const RetryWork = struct { var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; const scratch_dir = try std.fmt.bufPrint(&scratch_buf, "{s}/repair-scratch/failed-repo", .{self.retry.data_dir}); if (self.retry.stats) |s| _ = s.retry_attempts_total.fetchAdd(1, .monotonic); - _ = resync.resyncRepo(self.retry.allocator, self.retry.io, scratch_dir, self.retry.relay_http, did, self.retry.archive, self.store, &drops, &details) catch |err| { + _ = resync.resyncRepo(self.retry.allocator, self.retry.io, scratch_dir, self.retry.relay_http, did, self.retry.archive, self.store, &drops, &details, self.retry.trace_context) catch |err| { if (err == error.InvalidCommitRev) { if (self.retry.stats) |s| _ = s.backfill_drops.getPtr(.invalid_rev).fetchAdd(1, .monotonic); } diff --git a/src/internal/compact/steady.zig b/src/internal/compact/steady.zig index dc5f276..19ecb65 100644 --- a/src/internal/compact/steady.zig +++ b/src/internal/compact/steady.zig @@ -21,6 +21,7 @@ const std = @import("std"); const archive_mod = @import("../archive.zig"); const metrics = @import("../metrics.zig"); const meta_store = @import("../meta_store.zig"); +const observability = @import("../observability.zig"); const pass = @import("pass.zig"); const segment = @import("../segment.zig"); const segment_rewrite = @import("../segment_rewrite.zig"); @@ -50,6 +51,7 @@ pub const Compactor = struct { trigger: std.atomic.Value(bool) = .init(false), wake: Io.Event = .unset, stats: ?*metrics.Stats = null, + trace_context: observability.Context = observability.root_context, pub fn init(allocator: Allocator, io: Io, archive: *archive_mod.Archive, meta: *meta_store.Store, data_dir: []const u8) Compactor { return .{ @@ -171,19 +173,29 @@ pub const Compactor = struct { /// one force-rotate + pass + eviction; also the test seam pub fn runOnce(self: *Compactor) !pass.Stats { + var trace = observability.Span.start( + self.allocator, + "ingest/orchestrator", + "runDeleteCompaction", + self.trace_context, + &.{}, + ); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); const started = Io.Timestamp.now(self.io, .awake); - const result = self.runOnceInner() catch |err| { + const result = self.runOnceInner(trace.context()) catch |err| { if (self.stats) |s| s.observeCompactionPass(.error_result, elapsedUs(self.io, started)); - return err; + return @errorCast(err); }; if (self.stats) |s| s.observeCompactionPass(.ok, elapsedUs(self.io, started)); return result; } - fn runOnceInner(self: *Compactor) !pass.Stats { + fn runOnceInner(self: *Compactor, trace_context: observability.Context) !pass.Stats { // rows deleted while their segment was active get compacted this // pass; on_append ordering makes this race-free (module doc) - try self.archive.rotate(); + try self.archive.rotateTraced(trace_context); var wm = watermark.Store.init(self.meta); defer wm.deinit(self.io); diff --git a/src/internal/ingest.zig b/src/internal/ingest.zig index b95f620..6b66314 100644 --- a/src/internal/ingest.zig +++ b/src/internal/ingest.zig @@ -13,6 +13,7 @@ const segment = @import("segment.zig"); const cursor_mod = @import("cursor.zig"); const metrics = @import("metrics.zig"); const meta_store = @import("meta_store.zig"); +const observability = @import("observability.zig"); const pipeline_mod = @import("pipeline.zig"); const repair = @import("repair.zig"); const resync = @import("bootstrap/resync.zig"); @@ -177,18 +178,30 @@ pub const Consumer = struct { /// pipeline writer callback: the event was decoded once by the verify /// worker; its backing arena stays alive for the duration of this call. - pub fn emitVerified(ctx: *anyopaque, event: zat.FirehoseEvent, verdict: verify.Result) !void { + pub fn emitVerified( + ctx: *anyopaque, + event: zat.FirehoseEvent, + verdict: verify.Result, + trace_context: observability.Context, + ) !void { const self: *Consumer = @ptrCast(@alignCast(ctx)); - self.handleEvent(event, verdict, false) catch |err| { + self.handleEvent(event, verdict, false, trace_context) catch |err| { log.err("fatal event handling failure: {s}", .{@errorName(err)}); self.setFatal(err); return err; }; } - pub fn emitRepair(ctx: *anyopaque, completion: *repair.Completion) !void { + pub fn emitRepair(ctx: *anyopaque, completion: *repair.Completion, trace_context: observability.Context) !void { const self: *Consumer = @ptrCast(@alignCast(ctx)); const archive = self.archive orelse return error.RepairRequiresArchive; + const attributes = [_]observability.Attribute{ + .{ .key = "did", .value = .{ .string = completion.did } }, + }; + var trace = observability.Span.start(self.allocator, "ingest/backfill", "handleRepo", trace_context, &attributes); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit(); const alloc = arena.allocator(); @@ -204,6 +217,7 @@ pub const Consumer = struct { fetched_chain: verify.ChainUpdate, did: []const u8, upstream_seq: ?i64, + trace_context: observability.Context, stage_chain_on_flush: bool = false, fn stageChain(hook_ctx: *anyopaque, _: u64, last: u64) !void { @@ -227,9 +241,15 @@ pub const Consumer = struct { fn flush(sink: *@This()) !void { if (sink.batch.items.len == 0) return; const first = (if (sink.stage_chain_on_flush) - try sink.archive.appendBatchBeforeDurable(sink.batch.items, sink.now_us, sink, stageChain) + try sink.archive.appendBatchBeforeDurableTraced( + sink.batch.items, + sink.now_us, + sink, + stageChain, + sink.trace_context, + ) else - try sink.archive.appendBatch(sink.batch.items, sink.now_us)) orelse return; + try sink.archive.appendBatchTraced(sink.batch.items, sink.now_us, sink.trace_context)) orelse return; for (sink.batch.items, 0..) |row, i| { var stamped = row; stamped.seq = first + i; @@ -272,6 +292,7 @@ pub const Consumer = struct { .fetched_chain = completion.fetched_chain, .did = completion.did, .upstream_seq = completion.upstream_seq, + .trace_context = trace.context(), }; defer sink.batch.deinit(alloc); const sync_event = if (completion.sync_payload) |payload| @@ -291,8 +312,8 @@ pub const Consumer = struct { try writes.commitDurable(); self.drops.getPtr(.invalid_rev).* += 1; if (self.stats) |stats| _ = stats.drops.getPtr(.invalid_rev).fetchAdd(1, .monotonic); - for (completion.pending.items) |item| try self.handleEvent(item.event, item.verdict, true); - try archive.flushBlock(); + for (completion.pending.items) |item| try self.handleEvent(item.event, item.verdict, true, trace.context()); + try archive.flushBlockTraced(trace.context()); return; } @@ -318,8 +339,8 @@ pub const Consumer = struct { } sink.stage_chain_on_flush = true; try sink.flush(); - for (completion.pending.items) |item| try self.handleEvent(item.event, item.verdict, true); - try archive.flushBlock(); + for (completion.pending.items) |item| try self.handleEvent(item.event, item.verdict, true, trace.context()); + try archive.flushBlockTraced(trace.context()); } /// Atmos advances its cursor only after a non-empty delivery batch @@ -344,7 +365,13 @@ pub const Consumer = struct { if (self.fatal_error == null) self.fatal_error = err; } - fn handleEvent(self: *Consumer, event: zat.FirehoseEvent, verdict: verify.Result, replayed_after_repair: bool) !void { + fn handleEvent( + self: *Consumer, + event: zat.FirehoseEvent, + verdict: verify.Result, + replayed_after_repair: bool, + trace_context: observability.Context, + ) !void { var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit(); const alloc = arena.allocator(); @@ -404,7 +431,7 @@ pub const Consumer = struct { } var persisted: Persisted = .{}; const owns_cursor = !replayed_after_repair and self.pipeline == null; - if (self.archive) |a| persisted = self.persist(alloc, a, event, verdict, owns_cursor) catch |err| { + if (self.archive) |a| persisted = self.persist(alloc, a, event, verdict, owns_cursor, trace_context) catch |err| { // our own storage failing is fatal-worthy, not drop-worthy log.err("archive append failed: {s}", .{@errorName(err)}); return err; @@ -571,6 +598,7 @@ pub const Consumer = struct { event: zat.FirehoseEvent, verdict: verify.Result, stage_cursor: bool, + trace_context: observability.Context, ) !Persisted { const now_us: i64 = Io.Timestamp.now(self.io, .real).toMicroseconds(); @@ -665,7 +693,13 @@ pub const Consumer = struct { // upstream event advances the watermark beyond it. .stage_cursor = stage_cursor and event.seq() != null, }; - const first = (try a.appendBatchBeforeDurable(rows.items, now_us, &boundary, Boundary.stage)) orelse return .{}; + const first = (try a.appendBatchBeforeDurableTraced( + rows.items, + now_us, + &boundary, + Boundary.stage, + trace_context, + )) orelse return .{}; return .{ .first = first, .last = first + rows.items.len - 1, @@ -886,7 +920,7 @@ test "verifier silent drop waits for a later emitted cursor watermark" { .time = "2024-09-09T19:46:02.102Z", .ops = &.{}, } }; - try consumer.handleEvent(trigger, .{ .outcome = .needs_resync }, false); + try consumer.handleEvent(trigger, .{ .outcome = .needs_resync }, false, observability.root_context); try testing.expectEqual(@as(?i64, null), cursor.load(io)); const later: zat.FirehoseEvent = .{ .identity = .{ @@ -895,7 +929,7 @@ test "verifier silent drop waits for a later emitted cursor watermark" { .time = "2024-09-09T19:46:03.102Z", .handle = "later.test", } }; - try consumer.handleEvent(later, .{}, false); + try consumer.handleEvent(later, .{}, false, observability.root_context); try archive.flushBlock(); try testing.expectEqual(@as(?i64, 12), cursor.load(io)); } diff --git a/src/internal/observability.zig b/src/internal/observability.zig index 42021c1..43ae53e 100644 --- a/src/internal/observability.zig +++ b/src/internal/observability.zig @@ -254,6 +254,74 @@ pub fn tracer(scope: []const u8) !otel_api.trace.Tracer { return otel_api.getGlobalTracerProvider().getTracerWithScope(.{ .name = name }); } +pub const Context = []const otel_api.ContextKeyValue; +pub const root_context: Context = &.{}; +pub const Attribute = otel_api.AttributeKeyValue; + +/// One explicitly propagated operation span. Stream passes `context()` into +/// child operations rather than relying on thread-local state: bootstrap and +/// steady work cross worker threads, and pinned upstream propagates the Go +/// context across those same boundaries. +pub const Span = struct { + allocator: std.mem.Allocator, + value: otel_api.trace.Span = .{ .noop = .invalid }, + parent: Context = root_context, + owned_context: ?[]otel_api.ContextKeyValue = null, + ended: bool = false, + + pub fn start( + allocator: std.mem.Allocator, + scope: []const u8, + name: []const u8, + parent: Context, + attributes: []const otel_api.AttributeKeyValue, + ) Span { + var result: Span = .{ .allocator = allocator, .parent = parent }; + var span_tracer = tracer(scope) catch return result; + result.value = span_tracer.startSpan(name, .{ .attributes = attributes }, parent) catch return result; + const span_context = result.value.getSpanContext(); + if (span_context.isValid()) { + result.owned_context = otel_api.trace.trace_context.withActiveSpanContext( + allocator, + parent, + span_context, + ) catch null; + } + return result; + } + + pub fn context(self: *const Span) Context { + return if (self.owned_context) |owned| owned else self.parent; + } + + pub fn setAttribute(self: *Span, attribute: Attribute) void { + if (!self.ended) self.value.setAttribute(attribute); + } + + pub fn succeed(self: *Span) void { + if (self.ended) return; + self.value.setStatus(.{ .code = .ok }); + self.value.end(null); + self.ended = true; + } + + pub fn fail(self: *Span, err: anyerror) void { + if (self.ended) return; + self.value.recordException(err, null, null) catch {}; + self.value.setStatus(.{ .code = .@"error", .description = @errorName(err) }); + self.value.end(null); + self.ended = true; + } + + pub fn deinit(self: *Span) void { + if (!self.ended) self.succeed(); + if (self.owned_context) |owned| + otel_api.ContextKeyValue.deinitOwnedSlice(self.allocator, owned); + self.value.deinit(); + self.owned_context = null; + } +}; + fn envValue(entries: []const []const u8, key: []const u8) ?[]const u8 { for (entries) |entry| { const split = std.mem.indexOfScalar(u8, entry, '=') orelse entry.len; @@ -529,11 +597,16 @@ test "runtime exports a real protobuf span and resource during bounded shutdown" defer runtime.deinit(); try testing.expect(runtime.enabled()); - var span_tracer = try tracer("offline-contract"); - var span = try span_tracer.startSpan("provider-shutdown-span", .{}, &.{}); - span.setStatus(.{ .code = .ok }); - span.end(null); - span.deinit(); + var parent = Span.start(testing.allocator, "offline-contract", "provider-shutdown-span", root_context, &.{}); + const parent_context = otel_api.trace.trace_context.getActiveSpanContext(parent.context()).?; + var child = Span.start(testing.allocator, "offline-contract", "child-error-span", parent.context(), &.{}); + const child_context = otel_api.trace.trace_context.getActiveSpanContext(child.context()).?; + try testing.expectEqualSlices(u8, &parent_context.trace_id.bytes, &child_context.trace_id.bytes); + try testing.expect(!std.mem.eql(u8, &parent_context.span_id.bytes, &child_context.span_id.bytes)); + child.fail(error.OfflineContractFailure); + child.deinit(); + parent.succeed(); + parent.deinit(); try testing.expectEqual(otel_api.common.ProcessResult.success, runtime.shutdown(2_000)); try collector.await(io); @@ -541,6 +614,8 @@ test "runtime exports a real protobuf span and resource during bounded shutdown" try testing.expectEqualStrings("application/x-protobuf", capture.content_type[0..capture.content_type_len]); try testing.expectEqualStrings("offline-secret", capture.api_key[0..capture.api_key_len]); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "provider-shutdown-span") != null); + try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "child-error-span") != null); + try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "OfflineContractFailure") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "jetstream/offline-contract") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "stream-offline-test") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "v-test") != null); diff --git a/src/internal/pipeline.zig b/src/internal/pipeline.zig index a834226..9c9abb4 100644 --- a/src/internal/pipeline.zig +++ b/src/internal/pipeline.zig @@ -6,6 +6,7 @@ const std = @import("std"); const zat = @import("zat"); const verify = @import("verify.zig"); const repair = @import("repair.zig"); +const observability = @import("observability.zig"); const meta_store = @import("meta_store.zig"); const metrics = @import("metrics.zig"); @@ -49,8 +50,8 @@ pub const Pipeline = struct { /// Normal events are delivered in cross-DID verification completion /// order, while each DID remains FIFO. emit_ctx: *anyopaque, - emit: *const fn (*anyopaque, zat.FirehoseEvent, verify.Result) anyerror!void, - emit_repair: ?*const fn (*anyopaque, *repair.Completion) anyerror!void = null, + emit: *const fn (*anyopaque, zat.FirehoseEvent, verify.Result, observability.Context) anyerror!void, + emit_repair: ?*const fn (*anyopaque, *repair.Completion, observability.Context) anyerror!void = null, emit_cursor: ?*const fn (*anyopaque, i64) anyerror!void = null, mutex: Io.Mutex = .init, @@ -71,6 +72,7 @@ pub const Pipeline = struct { active_chains: usize = 0, writer_jobs_buffered: usize = 0, cursor_watermark: i64 = 0, + trace_context: observability.Context = observability.root_context, /// Highest positive relay seq accepted by this process. Atmos falls back /// to this global high-water when the inflight set is empty; using only /// the current completion batch can strand the cursor after a replayed @@ -453,9 +455,18 @@ pub const Pipeline = struct { var max_seq: i64 = 0; var normal_jobs: usize = 0; + var trace = observability.Span.start( + self.allocator, + "ingest/live", + "processBatch", + self.trace_context, + &.{}, + ); + defer trace.deinit(); for (batch.items, 0..) |delivery, i| switch (delivery) { .job => |job| { - if (job.event) |event| self.emit(self.emit_ctx, event, job.verdict) catch |err| { + if (job.event) |event| self.emit(self.emit_ctx, event, job.verdict, trace.context()) catch |err| { + trace.fail(err); release_from = i + 1; self.failJob(job, err); return; @@ -473,7 +484,8 @@ pub const Pipeline = struct { self.failPipeline(error.MissingRepairEmitter); return; }; - callback(self.emit_ctx, completion) catch |err| { + callback(self.emit_ctx, completion, trace.context()) catch |err| { + trace.fail(err); release_from = i + 1; completion.deinit(); self.finishBufferedRepair(); @@ -489,6 +501,7 @@ pub const Pipeline = struct { const watermark = self.cursorAfterBatch(); if (watermark > self.cursor_watermark) { if (self.emit_cursor) |callback| callback(self.emit_ctx, watermark) catch |err| { + trace.fail(err); self.failPipeline(err); return; }; @@ -496,6 +509,7 @@ pub const Pipeline = struct { } } self.finishWriterBatch(normal_jobs); + trace.succeed(); } } @@ -657,7 +671,7 @@ const Collector = struct { allocator: Allocator, // no lock needed: emit runs only on the pipeline's single writer thread - fn emit(ctx: *anyopaque, event: zat.FirehoseEvent, verdict: verify.Result) !void { + fn emit(ctx: *anyopaque, event: zat.FirehoseEvent, verdict: verify.Result, _: observability.Context) !void { _ = verdict; const self: *Collector = @ptrCast(@alignCast(ctx)); // test frames are #info envelopes carrying a name payload @@ -701,7 +715,7 @@ const SeqCollector = struct { mutex: Io.Mutex = .init, seqs: std.ArrayList(i64) = .empty, - fn emit(ctx: *anyopaque, event: zat.FirehoseEvent, _: verify.Result) !void { + fn emit(ctx: *anyopaque, event: zat.FirehoseEvent, _: verify.Result, _: observability.Context) !void { const self: *SeqCollector = @ptrCast(@alignCast(ctx)); const seq = event.seq() orelse return; self.mutex.lockUncancelable(self.io); @@ -720,7 +734,7 @@ const WakeCollector = struct { io: Io, emitted: Io.Event = .unset, - fn emit(ctx: *anyopaque, _: zat.FirehoseEvent, _: verify.Result) !void { + fn emit(ctx: *anyopaque, _: zat.FirehoseEvent, _: verify.Result, _: observability.Context) !void { const self: *WakeCollector = @ptrCast(@alignCast(ctx)); self.emitted.set(self.io); } @@ -730,7 +744,7 @@ const CursorProbe = struct { calls: usize = 0, last: i64 = 0, - fn emit(_: *anyopaque, _: zat.FirehoseEvent, _: verify.Result) !void {} + fn emit(_: *anyopaque, _: zat.FirehoseEvent, _: verify.Result, _: observability.Context) !void {} fn cursor(ctx: *anyopaque, value: i64) !void { const self: *CursorProbe = @ptrCast(@alignCast(ctx)); @@ -1136,7 +1150,7 @@ test "pipeline exposes emit failure without advancing emitted ticket" { defer threaded.deinit(); const Failing = struct { - fn emit(_: *anyopaque, _: zat.FirehoseEvent, _: verify.Result) !void { + fn emit(_: *anyopaque, _: zat.FirehoseEvent, _: verify.Result, _: observability.Context) !void { return error.PersistenceFailed; } }; diff --git a/src/internal/repair_integration_test.zig b/src/internal/repair_integration_test.zig index b6cee54..2dcb88b 100644 --- a/src/internal/repair_integration_test.zig +++ b/src/internal/repair_integration_test.zig @@ -85,7 +85,7 @@ test "repair completion durably emits tombstone then replacement offline" { .after_ticket = 0, .upstream_seq = 77, }; - try ingest.Consumer.emitRepair(&consumer, &completion); + try ingest.Consumer.emitRepair(&consumer, &completion, @import("observability.zig").root_context); try testing.expectEqual(@as(u64, 2), archive.committed_seq.load(.acquire)); try testing.expectEqual(@as(?i64, 77), cursor.load(io)); @@ -204,7 +204,7 @@ test "large repair keeps wire encoding scratch bounded" { .upstream_seq = 4096, }; - try ingest.Consumer.emitRepair(&consumer, &completion); + try ingest.Consumer.emitRepair(&consumer, &completion, @import("observability.zig").root_context); try testing.expectEqual(@as(u64, 4097), archive.committed_seq.load(.acquire)); try testing.expectEqual(@as(?i64, 4096), cursor.load(io)); try testing.expect(repair_alloc.total_requested_bytes < 256 * 1024); diff --git a/src/internal/timestamp/manager.zig b/src/internal/timestamp/manager.zig index c61b84f..14235a3 100644 --- a/src/internal/timestamp/manager.zig +++ b/src/internal/timestamp/manager.zig @@ -9,6 +9,7 @@ const std = @import("std"); const import_path = @import("import_path.zig"); const jobs_mod = @import("jobs.zig"); const runner = @import("runner.zig"); +const observability = @import("../observability.zig"); const Allocator = std.mem.Allocator; const Io = std.Io; @@ -137,6 +138,10 @@ pub const Manager = struct { self.wait(); } + pub fn setTraceContext(self: *Manager, trace_context: observability.Context) void { + self.runner_config.trace_context = trace_context; + } + fn ensureIdleLocked(self: *Manager) !void { if (self.active) return error.ImportJobInProgress; const current_id = try self.jobs.current() orelse return; diff --git a/src/internal/timestamp/runner.zig b/src/internal/timestamp/runner.zig index 7cb89b2..55217ea 100644 --- a/src/internal/timestamp/runner.zig +++ b/src/internal/timestamp/runner.zig @@ -14,6 +14,7 @@ const jobs_mod = @import("jobs.zig"); const parse = @import("parse.zig"); const rules_mod = @import("rules.zig"); const metrics = @import("../metrics.zig"); +const observability = @import("../observability.zig"); const Allocator = std.mem.Allocator; const Io = std.Io; @@ -30,6 +31,7 @@ pub const Config = struct { cancelled: ?*const std.atomic.Value(bool) = null, on_segment_applied_ctx: ?*anyopaque = null, on_segment_applied: ?*const fn (?*anyopaque, u64) void = null, + trace_context: observability.Context = observability.root_context, }; const Touched = struct { @@ -38,6 +40,16 @@ const Touched = struct { }; pub fn run(config: Config, id: []const u8) !void { + var trace = observability.Span.start( + config.allocator, + "ingest/orchestrator", + "RunImport", + config.trace_context, + &.{}, + ); + defer trace.deinit(); + defer trace.succeed(); + errdefer |trace_err| trace.fail(trace_err); const started_us = Io.Timestamp.now(config.io, .real).toMicroseconds(); defer if (config.stats) |stats| stats.import_phase.store(0, .release); var parsed = (try config.jobs.get(id)) orelse return error.ImportJobNotFound; @@ -45,10 +57,10 @@ pub fn run(config: Config, id: []const u8) !void { var record = parsed.value; if (record.state.terminal()) return error.ImportJobTerminal; - runInner(config, &record) catch |err| { + runInner(config, &record, trace.context()) catch |err| { if (err == error.ImportCancelled) { try config.jobs.put(record); - return err; + return @errorCast(err); } record.state = .failed; record.error_message = @errorName(err); @@ -56,7 +68,7 @@ pub fn run(config: Config, id: []const u8) !void { try config.jobs.put(record); observeTerminal(config, false, started_us, record.finished_at_us); cleanup(config, record.id) catch {}; - return err; + return @errorCast(err); }; record.state = .complete; @@ -68,7 +80,7 @@ pub fn run(config: Config, id: []const u8) !void { cleanup(config, record.id) catch {}; } -fn runInner(config: Config, record: *jobs_mod.Record) !void { +fn runInner(config: Config, record: *jobs_mod.Record, trace_context: observability.Context) !void { const job_path = try std.fs.path.join(config.allocator, &.{ config.scratch_root, record.id }); defer config.allocator.free(job_path); @@ -106,7 +118,7 @@ fn runInner(config: Config, record: *jobs_mod.Record) !void { // Rules are live before this boundary. Pre-existing active rows become // sealed and enter the catalog; later rows are stamped at append time. - config.archive.rotate() catch |err| { + config.archive.rotateTraced(trace_context) catch |err| { log.err("timestamp import {s}: archive rotate failed: {s}", .{ record.id, @errorName(err) }); return err; }; diff --git a/src/internal/tracing_integration_test.zig b/src/internal/tracing_integration_test.zig new file mode 100644 index 0000000..83b3de5 --- /dev/null +++ b/src/internal/tracing_integration_test.zig @@ -0,0 +1,113 @@ +const std = @import("std"); +const archive_mod = @import("archive.zig"); +const observability = @import("observability.zig"); +const segment = @import("segment.zig"); + +const Io = std.Io; + +const Capture = struct { + body: [128 * 1024]u8 = undefined, + len: usize = 0, +}; + +const Collector = struct { + fn receive(listener: *Io.net.Server, io: Io, capture: *Capture) !void { + var stream = try listener.accept(io); + defer stream.close(io); + var read_buffer: [4096]u8 = undefined; + var write_buffer: [4096]u8 = undefined; + var reader = stream.reader(io, &read_buffer); + var writer = stream.writer(io, &write_buffer); + var server = std.http.Server.init(&reader.interface, &writer.interface); + var request = try server.receiveHead(); + var body_buffer: [4096]u8 = undefined; + const body_reader = request.readerExpectNone(&body_buffer); + var body_writer = std.Io.Writer.fixed(&capture.body); + _ = try body_reader.streamRemaining(&body_writer); + capture.len = body_writer.buffered().len; + try request.respond("", .{ .status = .ok, .keep_alive = false }); + } +}; + +fn event(seq_hint: u64) segment.Event { + return .{ + .seq = 0, + .witnessed_at = @intCast(1_700_000_000_000_000 + seq_hint), + .indexed_at = 0, + .kind = .create, + .did = "did:plc:tracecontract", + .collection = "app.bsky.feed.post", + .rkey = "3l3qo2vuowo2b", + .rev = "3l3qo2vutsw2b", + .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", + }; +} + +test "real archive boundaries export the upstream span tree offline" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer listener.deinit(io); + var capture: Capture = .{}; + var collector = try io.concurrent(Collector.receive, .{ &listener, io, &capture }); + defer _ = collector.cancel(io) catch {}; + + var endpoint_buffer: [128]u8 = undefined; + const endpoint = try std.fmt.bufPrint( + &endpoint_buffer, + "OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:{d}", + .{listener.socket.address.getPort()}, + ); + var runtime = try observability.Runtime.init(testing.allocator, .{ + .io = io, + .environment = &.{ endpoint, "OTEL_BSP_SCHEDULE_DELAY=60000" }, + .service_name = "stream-tracing-contract", + .service_version = "offline", + }); + defer runtime.deinit(); + + var process_batch = observability.Span.start( + testing.allocator, + "ingest/live", + "processBatch", + observability.root_context, + &.{}, + ); + { + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buffer: [Io.Dir.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(io, &path_buffer); + var archive = try archive_mod.Archive.init(testing.allocator, io, path_buffer[0..path_len]); + defer archive.deinit(); + + archive.writer.max_events_per_block = 1; + archive.max_events_per_block = 1; + archive.max_segment_bytes = 1; + try archive.enableAsyncFlush(1); + var async_row = [_]segment.Event{event(1)}; + _ = try archive.appendBatchTraced(&async_row, 1, process_batch.context()); + + archive.disableAsyncFlush(); + archive.max_segment_bytes = archive_mod.default_max_segment_bytes; + var sync_row = [_]segment.Event{event(2)}; + _ = try archive.appendBatchTraced(&sync_row, 2, process_batch.context()); + } + process_batch.succeed(); + process_batch.deinit(); + + try testing.expectEqual(@import("otel-api").common.ProcessResult.success, runtime.shutdown(2_000)); + try collector.await(io); + const body = capture.body[0..capture.len]; + for ([_][]const u8{ + "processBatch", + "commitAsyncFlush", + "rotateIfFull", + "rotateLocked", + "flushAndRotateLocked", + "jetstream/ingest/live", + "jetstream/ingest", + }) |needle| try testing.expect(std.mem.indexOf(u8, body, needle) != null); +} diff --git a/src/main.zig b/src/main.zig index 760008c..4100ee4 100644 --- a/src/main.zig +++ b/src/main.zig @@ -879,11 +879,22 @@ pub fn main(init: std.process.Init.Minimal) !void { .shutdown_timeout_ms = shutdownTimeoutMillis(shutdown_timeout), }) catch |err| { log.err("setup tracing failed: {s}", .{@errorName(err)}); - return err; + return @errorCast(err); }; defer tracing.deinit(); if (tracing.enabled()) log.info("OTLP tracing enabled (service.name={s})", .{otel_service_name}); + var orchestrator_trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "Run", + observability.root_context, + &.{}, + ); + defer orchestrator_trace.deinit(); + defer orchestrator_trace.succeed(); + errdefer |err| orchestrator_trace.fail(err); + var hot_tail = tail_mod.Tail.init(allocator, io, subscribe_read_log_retention_bytes); defer hot_tail.deinit(); var stats: metrics.Stats = .{ @@ -916,7 +927,7 @@ pub fn main(init: std.process.Init.Minimal) !void { var archive = archive_mod.Archive.initWithStats(allocator, io, data_dir, &stats) catch |err| { logPersistenceFailure(data_dir, err); - return err; + return @errorCast(err); }; var archive_closed = false; defer { @@ -1014,14 +1025,14 @@ pub fn main(init: std.process.Init.Minimal) !void { const public_bind = if (public_addr_raw) |raw| resolveBindAddress(io, raw) catch |err| { log.err("invalid public bind address {s}: {s}", .{ raw, @errorName(err) }); - return err; + return @errorCast(err); } else BindAddress{ .address = .{ .ip4 = .unspecified(port) }, .port = port }; const debug_bind: ?BindAddress = if (debug_addr_raw) |raw| resolveBindAddress(io, raw) catch |err| { log.err("invalid debug bind address {s}: {s}", .{ raw, @errorName(err) }); - return err; + return @errorCast(err); } else null; @@ -1055,7 +1066,7 @@ pub fn main(init: std.process.Init.Minimal) !void { defer ws_server.deinit(); var listener = public_bind.address.listen(io, .{ .reuse_address = true }) catch |err| { log.err("failed to bind public listener {s}: {s}", .{ public_addr_raw orelse "legacy --port", @errorName(err) }); - return err; + return @errorCast(err); }; defer listener.deinit(io); var server_future = try io.concurrent(runWsServer, .{ &ws_server, &listener, &public_context }); @@ -1077,7 +1088,7 @@ pub fn main(init: std.process.Init.Minimal) !void { var debug_listener: ?Io.net.Server = if (debug_bind) |bind| bind.address.listen(io, .{ .reuse_address = true }) catch |err| { log.err("failed to bind debug listener {s}: {s}", .{ debug_addr_raw.?, @errorName(err) }); - return err; + return @errorCast(err); } else null; @@ -1110,7 +1121,7 @@ pub fn main(init: std.process.Init.Minimal) !void { // or --max-backfill-repos (debug) start one; serving is 503-gated // until the lifecycle commits steady_state. if (mid_lifecycle or bootstrap_requested) { - lifecycle.run(allocator, io, .{ + lifecycle.runTraced(allocator, io, .{ .data_dir = data_dir, .upstream = upstream, .relay_http = relay_http, @@ -1123,9 +1134,9 @@ pub fn main(init: std.process.Init.Minimal) !void { .max_inflight_bytes = backfill_max_inflight_bytes, .skip_merge_discovery = skip_merge_discovery, .compaction_enabled = compaction_interval_ns != 0, - }, &archive, &meta, &cursor_store, if (verifier) |*v| v else null, &stats) catch |err| { + }, &archive, &meta, &cursor_store, if (verifier) |*v| v else null, &stats, orchestrator_trace.context()) catch |err| { logPersistenceFailure(data_dir, err); - return err; + return @errorCast(err); }; hub.serving.store(true, .release); } @@ -1134,6 +1145,18 @@ pub fn main(init: std.process.Init.Minimal) !void { if (cursor) |c| log.info("resuming from persisted cursor {d}", .{c}); } + var steady_trace = observability.Span.start( + allocator, + "ingest/orchestrator", + "runSteadyState", + orchestrator_trace.context(), + &.{}, + ); + defer steady_trace.deinit(); + defer steady_trace.succeed(); + errdefer |err| steady_trace.fail(err); + import_manager.setTraceContext(steady_trace.context()); + // steady compactor: live tombstone set fed from the archive's append // hook, periodic + cap-triggered passes. interval 0 disables all of it // (run() returns immediately, no hook, no rebuild). @@ -1143,6 +1166,7 @@ pub fn main(init: std.process.Init.Minimal) !void { compactor.tombstone_cap = compaction_tombstone_cap; compactor.rewrite_workers = compaction_rewrite_workers; compactor.stats = &stats; + compactor.trace_context = steady_trace.context(); if (compaction_interval_ns != 0) { try compactor.rebuild(); archive.on_append_ctx = &compactor; @@ -1169,6 +1193,7 @@ pub fn main(init: std.process.Init.Minimal) !void { retry.host_workers = retry_host_workers; retry.max_delay_ns = retry_max_delay_ns; retry.stats = &stats; + retry.trace_context = steady_trace.context(); if (retry_interval_ns != 0) log.info("failed-repo retry loop on (interval {d}ns, workers {d}, host workers {d}, max delay {d}ns)", .{ retry_interval_ns, retry_workers, @@ -1214,6 +1239,7 @@ pub fn main(init: std.process.Init.Minimal) !void { .emit_repair = ingest.Consumer.emitRepair, .emit_cursor = ingest.Consumer.emitCursor, .cursor_watermark = cursor orelse 0, + .trace_context = steady_trace.context(), }; var repair_coordinator: ?repair_mod.Coordinator = if (verifier) |*v| .{ .allocator = allocator, @@ -1280,7 +1306,9 @@ pub fn main(init: std.process.Init.Minimal) !void { archive.close(); archive_closed = true; try listener_shutdowns.await(io); - if (shutdown_err) |err| return err; + if (shutdown_err) |err| return @errorCast(err); + steady_trace.succeed(); + orchestrator_trace.succeed(); try shutdownTracing(&tracing, shutdown_timeout); return; } @@ -1288,6 +1316,8 @@ pub fn main(init: std.process.Init.Minimal) !void { try pipe.drainAndStop(); archive.close(); archive_closed = true; + steady_trace.succeed(); + orchestrator_trace.succeed(); try shutdownTracing(&tracing, shutdown_timeout); } diff --git a/src/tests.zig b/src/tests.zig index 00b1095..abde642 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -37,6 +37,7 @@ comptime { _ = @import("internal/segment_fixture_test.zig"); _ = @import("internal/operational.zig"); _ = @import("internal/observability.zig"); + _ = @import("internal/tracing_integration_test.zig"); _ = @import("internal/inspect_all.zig"); _ = @import("internal/verify.zig"); _ = @import("internal/wire.zig");