From fdcfa0ef45d52f509fd3fd03c7863672825ff076 Mon Sep 17 00:00:00 2001 From: Noah Pederson Date: Fri, 9 Oct 2026 23:54:24 -0500 Subject: [PATCH] Add structured service diagnostics Expose replay causes, checkpoints, phases, and retry decisions so operators can diagnose stalled imports. Add periodic progress and debug request logs without headers or record bodies. Redact credentials and URL queries at the output boundary. Keep public error categories and retry behavior unchanged. --- cmd/appview/main.go | 7 ++ cmd/ui/main.go | 7 ++ docs/jetstream-replay.md | 3 + docs/logging.md | 56 +++++++++++++ internal/appview/http.go | 33 +++++--- internal/appview/indexing_test.go | 18 ++++ internal/appview/replay.go | 53 ++++++++++-- internal/appview/replay_logs.go | 46 ++++++++++ internal/appview/service.go | 11 ++- internal/httpx/http.go | 8 +- internal/logging/logging.go | 126 ++++++++++++++++++++++++++++ internal/logging/logging_test.go | 81 ++++++++++++++++++ internal/microcosm/replay_status.go | 52 ++++++++++-- internal/ui/http.go | 3 +- internal/upstream/queue.go | 66 +++++++++++++-- internal/upstream/queue_test.go | 40 +++++++++ nix/modules/appview.nix | 2 + nix/modules/common.nix | 11 +++ nix/modules/ui.nix | 2 + nix/services/provenance.nix | 2 + nix/tests/modules.nix | 3 + 21 files changed, 591 insertions(+), 39 deletions(-) create mode 100644 docs/logging.md create mode 100644 internal/appview/replay_logs.go create mode 100644 internal/logging/logging.go create mode 100644 internal/logging/logging_test.go diff --git a/cmd/appview/main.go b/cmd/appview/main.go index 975a103..0d402af 100644 --- a/cmd/appview/main.go +++ b/cmd/appview/main.go @@ -10,6 +10,7 @@ import ( "provenance/internal/config" "provenance/internal/domain" "provenance/internal/httpx" + "provenance/internal/logging" "provenance/internal/microcosm" reputationalg "provenance/internal/reputation/algorithms" "provenance/internal/storage" @@ -18,6 +19,10 @@ import ( ) func main() { + if err := logging.Setup("appview"); err != nil { + slog.Error("log configuration failed", "error", err) + os.Exit(1) + } if err := run(); err != nil { slog.Error("appview stopped", "error", err) os.Exit(1) @@ -29,11 +34,13 @@ func run() error { if err != nil { return err } + slog.Info("appview configuration loaded", "address", cfg.Address, "database", cfg.Database, "external_workers", cfg.ExternalWorkers, "soft_mode", cfg.SoftMode, "soft_vouches", cfg.SoftMode != domain.SoftDisabled, "soft_all_authors", domain.SoftAll(cfg.SoftMode), "soft_lists", domain.SoftLists(cfg.SoftMode), "replay", cfg.ReplayURL != "", "replay_origin", cfg.ReplayURL, "after_seq", cfg.ReplayRange.After, "before_seq", cfg.ReplayRange.Before, "trust_cache_ttl", cfg.TrustCacheTTL) store, err := storage.OpenWithSoft(cfg.Database, cfg.SoftMode) if err != nil { return err } defer store.Close() + slog.Info("database opened", "database", cfg.Database) ctx, cancel := httpx.Context() defer cancel() queue, err := upstream.New(ctx, cfg.ExternalWorkers) diff --git a/cmd/ui/main.go b/cmd/ui/main.go index 7e976ba..098e5ab 100644 --- a/cmd/ui/main.go +++ b/cmd/ui/main.go @@ -11,6 +11,7 @@ import ( "provenance/internal/config" "provenance/internal/domain" "provenance/internal/httpx" + "provenance/internal/logging" "provenance/internal/secure" "provenance/internal/storage" "provenance/internal/ui" @@ -20,6 +21,10 @@ import ( const PruneInterval = time.Hour func main() { + if err := logging.Setup("ui"); err != nil { + slog.Error("log configuration failed", "error", err) + os.Exit(1) + } if err := run(); err != nil { slog.Error("UI stopped", "error", err) os.Exit(1) @@ -31,11 +36,13 @@ func run() error { if err != nil { return err } + slog.Info("UI configuration loaded", "address", cfg.Address, "database", cfg.Database, "external_workers", cfg.ExternalWorkers, "origin", cfg.Origin, "appview_url", cfg.AppviewURL) store, err := storage.Open(cfg.Database) if err != nil { return err } defer store.Close() + slog.Info("database opened", "database", cfg.Database) vault, err := secure.New(store, cfg.EncryptionKey) if err != nil { return err diff --git a/docs/jetstream-replay.md b/docs/jetstream-replay.md index fdef5ba..9a43655 100644 --- a/docs/jetstream-replay.md +++ b/docs/jetstream-replay.md @@ -76,6 +76,9 @@ A successful sweep commits the end checkpoint, including an empty filtered tail. ## Status and metrics +Structured logs report replay phases, periodic progress, retry decisions, and terminal error causes. +See [Service logs](logging.md) for debug configuration and field definitions. + `/healthz` reports process health. `/readyz` reports replay readiness separately. Replay readiness requires a live connection and activity within two minutes. The last committed event must also remain within two minutes of wall time. diff --git a/docs/logging.md b/docs/logging.md new file mode 100644 index 0000000..c18ec4e --- /dev/null +++ b/docs/logging.md @@ -0,0 +1,56 @@ +# Service logs + +Both services emit one JSON object per line to stderr through `slog`. +Each record includes a timestamp, level, message, and service name. + +`PROVENANCE_LOG_LEVEL` selects `debug`, `info`, `warn`, or `error`. +The default is `info`. An invalid level stops service startup. + +## Debug configuration + +Add this assignment to your private environment or `.env`: + +```dotenv +PROVENANCE_LOG_LEVEL=debug +``` + +Stop the development launcher. +Start the launcher with its updated environment: + +```sh +direnv exec . nix run .#services +``` + +NixOS exposes `services.provenance.appview.logLevel` and `services.provenance.ui.logLevel`. +The development service module exposes `logLevel` and accepts the environment override. + +## Log levels + +| Level | Records | +| --- | --- | +| Info | Safe startup configuration, database open, replay attempts, phase changes, archive plans, segment starts, completion, and shutdown | +| Info | Replay progress once per minute, including a checkpoint and byte/event deltas during an idle download | +| Warn | Upstream request/read errors, failed HTTP statuses, cooldown-related retries, and stream reconnect decisions | +| Error | Terminal replay errors with their causes, failed phase, and durable checkpoint | +| Debug | HTTP request duration, external queue wait, response status/length, bytes read, archive segment/block/range identifiers, redirects, and committed batches | + +Duration and backoff fields use nanoseconds. Byte fields use bytes. +Replay counters in logs cover the current worker run. The checkpoint persists in SQLite. +A zero checkpoint delta with a positive byte delta describes a download before its first batch commit. + +Terminal errors include an `error.message` and the wrapped `error.types`. +The message preserves causes that an indexing wrapper otherwise hides. +Public status responses and metrics retain sanitized error categories only. + +## Data restrictions + +Request logs omit authorization headers, cookies, URL queries, fragments, bodies, and record contents. +Diagnostic fields include HTTP ranges, response lengths, and retry timing. +The output handler redacts configured credentials, bearer tokens, credential assignments, and sensitive attributes. +Archive identifiers come from a fixed query-parameter allowlist, not the complete query string. +Error fields have a 4096-byte limit after redaction. +Error messages retain provider diagnostics after redaction. +Before you share logs outside the operator team, review them. + +Debug logs can produce substantial output during an archive import. +Use debug level for diagnosis. Then restore info level. diff --git a/internal/appview/http.go b/internal/appview/http.go index 7fb4819..3d2ef15 100644 --- a/internal/appview/http.go +++ b/internal/appview/http.go @@ -13,6 +13,7 @@ import ( "provenance/internal/atproto" "provenance/internal/domain" "provenance/internal/httpx" + "provenance/internal/logging" "provenance/internal/trust" ) @@ -123,7 +124,7 @@ func (h *HTTP) reputation(w http.ResponseWriter, r *http.Request) { tag, mode := query(r) result, err := h.service.Reputation(r.Context(), r.URL.Query().Get("source"), r.URL.Query().Get("target"), tag, mode, options) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } if r.URL.Query().Get("include_path") != "true" { @@ -149,7 +150,7 @@ func (h *HTTP) evaluate(r *http.Request) (Evaluation, error) { func (h *HTTP) relative(w http.ResponseWriter, r *http.Request) { result, err := h.evaluate(r) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } if !result.Trusted && r.URL.Query().Get("include_path") != "true" { @@ -162,7 +163,7 @@ func (h *HTTP) relative(w http.ResponseWriter, r *http.Request) { func (h *HTTP) trusted(w http.ResponseWriter, r *http.Request) { result, err := h.evaluate(r) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]bool{"trusted": result.Trusted}) @@ -185,7 +186,7 @@ func (h *HTTP) aggregate(w http.ResponseWriter, r *http.Request) { source = r.Header.Get(IdentityHeader) p, err := h.service.Config(r.Context(), source) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } threshold = p.Threshold @@ -203,7 +204,7 @@ func (h *HTTP) aggregate(w http.ResponseWriter, r *http.Request) { tag, mode := query(r) result, err := h.service.RelativeReputation(r.Context(), source, values.Get("target"), tag, mode, threshold, options) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, result) @@ -212,7 +213,7 @@ func (h *HTTP) aggregate(w http.ResponseWriter, r *http.Request) { func (h *HTTP) tags(w http.ResponseWriter, r *http.Request) { tags, err := h.service.Tags(r.Context(), r.URL.Query().Get("prefix")) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, tags) @@ -230,7 +231,7 @@ func (h *HTTP) vouches(w http.ResponseWriter, r *http.Request) { } vouches, err := h.service.Vouches(r.Context(), r.URL.Query().Get("author"), offset) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, vouches) @@ -238,7 +239,7 @@ func (h *HTTP) vouches(w http.ResponseWriter, r *http.Request) { func (h *HTTP) join(w http.ResponseWriter, r *http.Request) { if err := h.service.Join(r.Context(), r.Header.Get(IdentityHeader)); err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]bool{"joined": true}) @@ -246,7 +247,7 @@ func (h *HTTP) join(w http.ResponseWriter, r *http.Request) { func (h *HTTP) sync(w http.ResponseWriter, r *http.Request) { if err := h.service.Sync(r.Context(), r.Header.Get(IdentityHeader)); err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]bool{"synced": true}) @@ -255,7 +256,7 @@ func (h *HTTP) sync(w http.ResponseWriter, r *http.Request) { func (h *HTTP) config(w http.ResponseWriter, r *http.Request) { p, err := h.service.Config(r.Context(), r.Header.Get(IdentityHeader)) if err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, p) @@ -274,13 +275,20 @@ func (h *HTTP) threshold(w http.ResponseWriter, r *http.Request) { return } if err := h.service.SetThreshold(r.Context(), r.Header.Get(IdentityHeader), *body.Threshold); err != nil { - h.fail(w, err) + h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]float64{"threshold": *body.Threshold}) } -func (h *HTTP) fail(w http.ResponseWriter, err error) { +func (h *HTTP) fail(w http.ResponseWriter, r *http.Request, err error) { + status := http.StatusBadRequest + if errors.Is(err, trust.ErrLimit) { + status = http.StatusServiceUnavailable + } else if errors.Is(err, sql.ErrNoRows) { + status = http.StatusNotFound + } + slog.Warn("appview request failed", "component", "http", "method", r.Method, "path", r.URL.Path, "status", status, "error", logging.Error(err)) if errors.Is(err, trust.ErrLimit) { httpx.Error(w, http.StatusServiceUnavailable, err.Error()) return @@ -289,6 +297,5 @@ func (h *HTTP) fail(w http.ResponseWriter, err error) { httpx.Error(w, http.StatusNotFound, "participant does not exist") return } - slog.Warn("appview request failed", "error", err) httpx.Error(w, http.StatusBadRequest, "request failed. Verify the identifiers and tag mode, or retry after the index sync.") } diff --git a/internal/appview/indexing_test.go b/internal/appview/indexing_test.go index eacda17..9ccc19e 100644 --- a/internal/appview/indexing_test.go +++ b/internal/appview/indexing_test.go @@ -1,8 +1,11 @@ package appview import ( + "bytes" "context" "errors" + "fmt" + "log/slog" "net/http" "net/http/httptest" "strings" @@ -49,6 +52,21 @@ func TestIndexPermanentFailure(t *testing.T) { } } +func TestReplayFailureLog(t *testing.T) { + var output bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&output, nil))) + t.Cleanup(func() { slog.SetDefault(previous) }) + s, _ := algorithmService(t, algorithms.Optimistic{}) + failure := domain.FailIndex(domain.IndexUpstream, fmt.Errorf("archive block decode: %w", errors.New("checksum mismatch"))) + s.RunReplay(context.Background(), &failedReplay{err: failure}) + for _, value := range []string{"checksum mismatch", `"checkpoint":0`, `"source":"failed-replay"`, `"phase":"failed"`} { + if !strings.Contains(output.String(), value) { + t.Errorf("missing failure context: %s", value) + } + } +} + func TestIndexStatusAuthorization(t *testing.T) { s, _ := algorithmService(t, algorithms.Optimistic{}) r := httptest.NewRecorder() diff --git a/internal/appview/replay.go b/internal/appview/replay.go index 4c6d73b..4afe3af 100644 --- a/internal/appview/replay.go +++ b/internal/appview/replay.go @@ -3,10 +3,12 @@ package appview import ( "context" "errors" + "fmt" "log/slog" "time" "provenance/internal/domain" + "provenance/internal/logging" "provenance/internal/upstream" ) @@ -19,17 +21,28 @@ type ReplayStream interface { // Use sequence checkpoints only for v2. Legacy streams retain their timestamp cursors. func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { delay := InitialRetryDelay + workerStarted := time.Now() + logger := slog.Default().With("component", "replay", "source", stream.Source(), "after_seq", stream.Range().After, "before_seq", stream.Range().Before) + var stats replayLogStats + stopLogs := s.replayLogs(ctx, logger, &stats) + defer func() { + stopLogs() + status := s.index.snapshot(s.now()) + logger.Info("replay worker stopped", "phase", status.Phase, "checkpoint", status.Checkpoint, "events", stats.events.Load(), "batches", stats.batches.Load(), "download_bytes", stats.bytes.Load(), "duration", time.Since(workerStarted)) + }() s.index.mu.Lock() s.index.state.Range = stream.Range() s.index.mu.Unlock() s.index.phase(domain.IndexStarting, s.now()) + logger.Info("replay started", "soft_mode", s.softMode) if err := s.BootstrapPending(ctx); err != nil { - slog.Warn("pending bootstrap failed", "error", err) + logger.Warn("pending bootstrap failed", "error", logging.Error(err)) } + attempt := 0 for ctx.Err() == nil { cursor, err := s.store.Cursor(ctx, stream.Source(), stream.Range().After) if err != nil { - s.indexFailure(domain.IndexStorage) + s.indexFailure(domain.IndexStorage, err, logger) return } s.index.mu.Lock() @@ -37,6 +50,8 @@ func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { s.index.mu.Unlock() processed := 0 started := time.Now() + attempt++ + logger.Info("replay attempt started", "attempt", attempt, "checkpoint", cursor) if err == nil { err = stream.ConsumeReplay(ctx, cursor, s.wanted, func(ctx context.Context, changes []domain.Change, cursor int64) error { batchStarted := time.Now() @@ -44,16 +59,34 @@ func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { defer s.syncMu.Unlock() for index := range changes { if err := s.prepareChange(ctx, &changes[index]); err != nil { - return domain.FailIndex(domain.IndexEvent, err) + return domain.FailIndex(domain.IndexEvent, fmt.Errorf("prepare sequence %d: %w", changes[index].Sequence, err)) } } if err := s.store.ApplyReplay(ctx, stream.Source(), changes, cursor); err != nil { - return domain.FailIndex(domain.IndexStorage, err) + return domain.FailIndex(domain.IndexStorage, fmt.Errorf("commit batch ending at sequence %d: %w", cursor, err)) } processed += len(changes) s.index.commit(changes, cursor, s.now(), time.Since(batchStarted)) + stats.events.Add(int64(len(changes))) + stats.batches.Add(1) + logger.Debug("replay batch committed", "checkpoint", cursor, "events", len(changes), "duration", time.Since(batchStarted)) return nil - }, func(progress domain.ReplayProgress) { s.index.progress(progress, s.now()) }) + }, func(progress domain.ReplayProgress) { + var previous domain.IndexPhase + if progress.Phase != "" { + previous = s.index.snapshot(s.now()).Phase + } + if progress.Bytes > 0 { + stats.bytes.Add(progress.Bytes) + } + s.index.progress(progress, s.now()) + if progress.Phase != "" && progress.Phase != previous { + logger.Info("replay phase changed", "phase", progress.Phase, "checkpoint", s.index.snapshot(s.now()).Checkpoint) + } + if progress.Tip > 0 { + logger.Info("archive plan ready", "archive_tip", minTip(progress.Tip, stream.Range().Before)) + } + }) } if ctx.Err() != nil { s.index.phase(domain.IndexStopped, s.now()) @@ -61,6 +94,7 @@ func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { } if err == nil && stream.Range().Before > 0 { s.index.phase(domain.IndexComplete, s.now()) + logger.Info("replay complete", "checkpoint", s.index.snapshot(s.now()).Checkpoint, "events", stats.events.Load(), "batches", stats.batches.Load(), "download_bytes", stats.bytes.Load(), "duration", time.Since(workerStarted)) return } var failure *domain.IndexError @@ -69,7 +103,7 @@ func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { if failure != nil { kind = failure.Kind() } - s.indexFailure(kind) + s.indexFailure(kind, err, logger) return } s.index.failures.WithLabelValues(string(domain.IndexTransient)).Inc() @@ -77,7 +111,7 @@ func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { s.index.phase(domain.IndexRetry, s.now()) delay = reconnectDelay(delay, processed, time.Since(started)) pause := upstream.Jitter(delay) - slog.Warn("replay stream disconnected", "kind", domain.IndexTransient, "retry", pause) + logger.Warn("replay reconnect scheduled", "kind", domain.IndexTransient, "attempt", attempt, "checkpoint", s.index.snapshot(s.now()).Checkpoint, "events", processed, "duration", time.Since(started), "retry_after", pause, "error", logging.Error(err)) if err := upstream.Wait(ctx, pause); err != nil { s.index.phase(domain.IndexStopped, s.now()) return @@ -87,11 +121,12 @@ func (s *Service) RunReplay(ctx context.Context, stream ReplayStream) { s.index.phase(domain.IndexStopped, s.now()) } -func (s *Service) indexFailure(kind domain.IndexFailureKind) { +func (s *Service) indexFailure(kind domain.IndexFailureKind, err error, logger *slog.Logger) { + previous := s.index.snapshot(s.now()) s.index.mu.Lock() s.index.state.Failure = kind s.index.mu.Unlock() s.index.phase(domain.IndexFailed, s.now()) s.index.failures.WithLabelValues(string(kind)).Inc() - slog.Error("indexing stopped; repair the cause and restart the service", "kind", kind) + logger.Error("replay stopped; repair the cause and restart the service", "kind", kind, "phase", domain.IndexFailed, "failed_phase", previous.Phase, "checkpoint", previous.Checkpoint, "error", logging.Error(err)) } diff --git a/internal/appview/replay_logs.go b/internal/appview/replay_logs.go new file mode 100644 index 0000000..3e86933 --- /dev/null +++ b/internal/appview/replay_logs.go @@ -0,0 +1,46 @@ +package appview + +import ( + "context" + "log/slog" + "sync/atomic" + "time" +) + +const replayLogInterval = time.Minute + +type replayLogStats struct{ bytes, events, batches atomic.Int64 } + +// Report durable progress even when the SDK waits on a download and emits no batches. +func (s *Service) replayLogs(ctx context.Context, logger *slog.Logger, stats *replayLogStats) func() { + ctx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(replayLogInterval) + defer ticker.Stop() + previous := s.index.snapshot(s.now()).Checkpoint + var previousBytes, previousEvents int64 + started := s.now() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + status := s.index.snapshot(s.now()) + bytes, events := stats.bytes.Load(), stats.events.Load() + logger.Info("replay progress", "phase", status.Phase, "checkpoint", status.Checkpoint, "checkpoint_delta", status.Checkpoint-previous, "archive_tip", status.ArchiveTip, "download_bytes", bytes, "byte_delta", bytes-previousBytes, "events", events, "event_delta", events-previousEvents, "batches", stats.batches.Load(), "last_commit", status.LastCommit, "last_event", status.LastEvent, "elapsed", s.now().Sub(started)) + previous = status.Checkpoint + previousBytes, previousEvents = bytes, events + } + } + }() + return func() { cancel(); <-done } +} + +func minTip(tip, before int64) int64 { + if before > 0 { + return min(tip, before) + } + return tip +} diff --git a/internal/appview/service.go b/internal/appview/service.go index 4b454fb..19ab74e 100644 --- a/internal/appview/service.go +++ b/internal/appview/service.go @@ -14,6 +14,7 @@ import ( "provenance/internal/atproto" "provenance/internal/domain" + "provenance/internal/logging" "provenance/internal/reputation" reputationalg "provenance/internal/reputation/algorithms" "provenance/internal/storage" @@ -439,17 +440,23 @@ func (s *Service) wanted(ctx context.Context, did string, stamp int64) (bool, er // Reconnect to the event source, never to each participant's PDS. func (s *Service) Run(ctx context.Context, stream Stream) { delay := InitialRetryDelay + logger := slog.Default().With("component", "legacy_stream", "source", s.streamSource(), "soft_mode", s.softMode) + logger.Info("legacy stream started") + defer logger.Info("legacy stream worker stopped") prepared := false + attempt := 0 for { cursor, err := s.store.Cursor(ctx, s.streamSource(), time.Now().UnixMicro()) if err == nil && !prepared { if err := s.BootstrapPending(ctx); err != nil { - slog.Warn("pending bootstrap failed", "error", err) + logger.Warn("pending bootstrap failed", "error", logging.Error(err)) } prepared = true } processed := 0 started := time.Now() + attempt++ + logger.Info("legacy stream attempt started", "attempt", attempt, "cursor_time_us", cursor) if err == nil { err = stream.Consume(ctx, cursor, s.wanted, func(ctx context.Context, change domain.Change) error { if err := s.Apply(ctx, change); err != nil { @@ -464,7 +471,7 @@ func (s *Service) Run(ctx context.Context, stream Stream) { } delay = reconnectDelay(delay, processed, time.Since(started)) pause := upstream.Jitter(delay) - slog.Warn("event stream disconnected", "error", err, "retry", pause) + logger.Warn("legacy stream reconnect scheduled", "attempt", attempt, "events", processed, "duration", time.Since(started), "retry_after", pause, "error", logging.Error(err)) if err := upstream.Wait(ctx, pause); err != nil { return } diff --git a/internal/httpx/http.go b/internal/httpx/http.go index a9556f9..63036fd 100644 --- a/internal/httpx/http.go +++ b/internal/httpx/http.go @@ -49,6 +49,10 @@ func Decode(w http.ResponseWriter, r *http.Request, value any) error { func Headers(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + started := time.Now() + logger := slog.Default().With("component", "http", "method", r.Method, "path", r.URL.Path) + logger.Debug("HTTP request started") + defer func() { logger.Debug("HTTP request finished", "duration", time.Since(started)) }() w.Header().Set("X-Content-Type-Options", "nosniff") // Preserve form origins without sending referrers to other origins. w.Header().Set("Referrer-Policy", "same-origin") @@ -65,6 +69,7 @@ func Serve(ctx context.Context, address string, handler http.Handler) error { server := &http.Server{Addr: address, Handler: Headers(handler), ReadHeaderTimeout: ReadTimeout, ReadTimeout: ReadTimeout, WriteTimeout: WriteTimeout, IdleTimeout: IdleTimeout, MaxHeaderBytes: MaxBodyBytes} go func() { <-ctx.Done() + slog.Info("HTTP shutdown started", "address", address) shutdown, cancel := context.WithTimeout(context.Background(), ShutdownTimeout) defer cancel() if err := server.Shutdown(shutdown); err != nil { @@ -72,9 +77,10 @@ func Serve(ctx context.Context, address string, handler http.Handler) error { server.Close() } }() - slog.Info("service ready", "address", address) + slog.Info("HTTP server starts", "address", address) err := server.ListenAndServe() if errors.Is(err, http.ErrServerClosed) { + slog.Info("HTTP server stopped", "address", address) return nil } return err diff --git a/internal/logging/logging.go b/internal/logging/logging.go new file mode 100644 index 0000000..fd978ab --- /dev/null +++ b/internal/logging/logging.go @@ -0,0 +1,126 @@ +package logging + +import ( + "errors" + "fmt" + "io" + "log/slog" + "net/url" + "os" + "regexp" + "strings" +) + +const ( + maxErrorDepth = 16 + maxFieldBytes = 4096 + redacted = "[redacted]" +) + +var ( + urlPattern = regexp.MustCompile(`https?://[^\s"'<>]+`) + bearerPattern = regexp.MustCompile(`(?i)\bBearer\s+[^\s"',;]+`) + secretPattern = regexp.MustCompile(`(?i)\b((?:access_token|refresh_token|accessJwt|refreshJwt|api_key|service_key|client_secret|encryption_key|password|authorization|cookie)["']?\s*[:=]\s*["']?)[^\s"',;}]+`) +) + +func Setup(service string) error { + level := slog.LevelInfo + switch strings.ToLower(os.Getenv("PROVENANCE_LOG_LEVEL")) { + case "", "info": + case "debug": + level = slog.LevelDebug + case "warn": + level = slog.LevelWarn + case "error": + level = slog.LevelError + default: + return errors.New("PROVENANCE_LOG_LEVEL must be debug, info, warn, or error") + } + secrets := []string{os.Getenv("PROVENANCE_SERVICE_KEY"), os.Getenv("PROVENANCE_ENCRYPTION_KEY"), os.Getenv("JETSTREAM_API_KEY")} + slog.SetDefault(New(os.Stderr, level, service, secrets)) + return nil +} + +// Redact shared credentials and URL queries at the output boundary. +func New(w io.Writer, level slog.Level, service string, secrets []string) *slog.Logger { + secrets = append([]string(nil), secrets...) + clean := func(value string) string { return sanitize(value, secrets) } + handler := slog.NewJSONHandler(w, &slog.HandlerOptions{Level: level, ReplaceAttr: func(_ []string, attr slog.Attr) slog.Attr { + if sensitive(attr.Key) { + return slog.String(attr.Key, redacted) + } + if err, ok := attr.Value.Any().(error); ok { + attr.Value = Error(err) + } + if attr.Value.Kind() == slog.KindString { + attr.Value = slog.StringValue(clean(attr.Value.String())) + } + return attr + }}) + return slog.New(handler).With("service", service) +} + +// Preserve causes hidden by wrappers such as domain.IndexError. Keep public error responses unchanged. +func Error(err error) slog.Value { + if err == nil { + return slog.GroupValue() + } + var messages, types []string + pending := []error{err} + for depth := 0; len(pending) > 0 && depth < maxErrorDepth; depth++ { + current := pending[len(pending)-1] + pending = pending[:len(pending)-1] + if current == nil { + continue + } + message := current.Error() + if !strings.Contains(strings.Join(messages, ": "), message) { + messages = append(messages, message) + } + types = append(types, fmt.Sprintf("%T", current)) + switch wrapped := current.(type) { + case interface{ Unwrap() []error }: + pending = append(pending, wrapped.Unwrap()...) + case interface{ Unwrap() error }: + pending = append(pending, wrapped.Unwrap()) + } + } + return slog.GroupValue(slog.String("message", strings.Join(messages, ": ")), slog.Any("types", types)) +} + +func sensitive(key string) bool { + switch strings.ToLower(key) { + case "authorization", "cookie", "set-cookie", "password", "secret", "api_key", "service_key", "encryption_key", "access_token", "refresh_token", "accessjwt", "refreshjwt", "client_secret": + return true + } + return false +} + +func sanitize(value string, secrets []string) string { + for _, secret := range secrets { + if secret == "" { + continue + } + for _, encoded := range []string{secret, url.QueryEscape(secret), url.PathEscape(secret)} { + value = strings.ReplaceAll(value, encoded, redacted) + } + } + value = bearerPattern.ReplaceAllString(value, "Bearer "+redacted) + value = secretPattern.ReplaceAllString(value, "${1}"+redacted) + value = urlPattern.ReplaceAllStringFunc(value, func(raw string) string { + parsed, err := url.Parse(raw) + if err != nil { + if !strings.ContainsAny(raw, "?#@") { + return raw + } + return "[redacted_url]" + } + parsed.User, parsed.RawQuery, parsed.Fragment = nil, "", "" + parsed.ForceQuery = false + return parsed.String() + }) + if len(value) > maxFieldBytes { + value = value[:maxFieldBytes] + "..." + } + return value +} diff --git a/internal/logging/logging_test.go b/internal/logging/logging_test.go new file mode 100644 index 0000000..02ae926 --- /dev/null +++ b/internal/logging/logging_test.go @@ -0,0 +1,81 @@ +package logging + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "strings" + "testing" + + "provenance/internal/domain" +) + +func TestErrorRedaction(t *testing.T) { + var output bytes.Buffer + const key = "test-key-with/special+characters" + logger := New(&output, slog.LevelDebug, "appview", []string{key}) + err := domain.FailIndex(domain.IndexUpstream, fmt.Errorf("getBlock: %w", errors.New(`HTTP 503 https://user:password@archive.example/getBlock?token=query-secret#fragment Bearer bearer-secret access_token=token-secret `+key))) + logger.Error("replay failed", "error", err, "api_key", key) + logger.Error("replay failed", "error", Error(err)) + for _, secret := range []string{key, "query-secret", "bearer-secret", "token-secret", "password@", "#fragment"} { + if strings.Contains(output.String(), secret) { + t.Fatalf("log exposed %s", secret) + } + } + for _, detail := range []string{"HTTP 503", "archive.example/getBlock", "getBlock", "domain.IndexError", "appview"} { + if !strings.Contains(output.String(), detail) { + t.Errorf("missing error detail: %s", detail) + } + } + decoder := json.NewDecoder(&output) + for range 2 { + var record map[string]any + if err := decoder.Decode(&record); err != nil { + t.Fatal(err) + } + if _, ok := record["error"].(map[string]any); !ok { + t.Fatal("error lacks structured context") + } + } +} + +func TestLogLevel(t *testing.T) { + var output bytes.Buffer + logger := New(&output, slog.LevelInfo, "appview", nil) + logger.Debug("batch committed") + logger.Info("replay complete") + if strings.Contains(output.String(), "batch committed") || !strings.Contains(output.String(), "replay complete") { + t.Fatal("log level filter failed") + } +} + +func TestSetupLevel(t *testing.T) { + previous := slog.Default() + t.Cleanup(func() { slog.SetDefault(previous) }) + for _, level := range []string{"debug", "info", "warn", "error"} { + t.Setenv("PROVENANCE_LOG_LEVEL", level) + if err := Setup("appview"); err != nil { + t.Fatal(err) + } + if slog.Default().Enabled(context.Background(), slog.LevelDebug) != (level == "debug") { + t.Fatalf("incorrect level: %s", level) + } + } + t.Setenv("PROVENANCE_LOG_LEVEL", "invalid") + if err := Setup("appview"); err == nil { + t.Fatal("invalid log level accepted") + } +} + +func TestFieldLimit(t *testing.T) { + var output bytes.Buffer + const key = "secret-value" + logger := New(&output, slog.LevelInfo, "appview", []string{key}) + logger.Error("replay failed", "error", errors.New(strings.Repeat("x", maxFieldBytes-3)+key)) + if strings.Contains(output.String(), "sec") { + t.Fatal("field truncation exposed part of a secret") + } +} diff --git a/internal/microcosm/replay_status.go b/internal/microcosm/replay_status.go index 4572eb5..0f70c4e 100644 --- a/internal/microcosm/replay_status.go +++ b/internal/microcosm/replay_status.go @@ -6,12 +6,16 @@ import ( "encoding/json" "errors" "io" + "log/slog" "net/http" + "strings" + "time" "github.com/coder/websocket" "github.com/jcalabro/atmos/xrpc" "provenance/internal/domain" + "provenance/internal/logging" ) const replayRedirectLimit = 10 @@ -29,6 +33,7 @@ func replayRedirect(req *http.Request, via []*http.Request) error { if req.URL.Scheme != origin.Scheme || req.URL.Host != origin.Host { req.Header.Del("Authorization") } + slog.Debug("archive redirect", "component", "archive_http", "from_host", via[len(via)-1].URL.Host, "to_host", req.URL.Host, "path", req.URL.Path) return nil } @@ -39,10 +44,30 @@ type replayTransport struct { } func (t *replayTransport) RoundTrip(req *http.Request) (*http.Response, error) { + segment := req.URL.Query().Get("segment") + if segment == "" { + segment = req.URL.Query().Get("name") + } + logger := slog.Default().With("component", "archive_http", "operation", req.URL.Path, "host", req.URL.Host, "segment", segment, "block_index", req.URL.Query().Get("blockIndex"), "range", req.Header.Get("Range")) + started := time.Now() + logger.Debug("archive request started") + if strings.HasPrefix(req.Header.Get("Range"), "bytes=0-") && segment != "" { + logger.Info("archive segment download started") + } resp, err := t.base.RoundTrip(req) if err != nil { + level := slog.LevelWarn + if errors.Is(err, context.Canceled) { + level = slog.LevelDebug + } + logger.Log(req.Context(), level, "archive request failed", "duration", time.Since(started), "error", logging.Error(err)) return nil, err } + logger = logger.With("status", resp.StatusCode) + logger.Debug("archive response received", "header_duration", time.Since(started), "content_length", resp.ContentLength, "content_range", resp.Header.Get("Content-Range")) + if resp.StatusCode >= http.StatusBadRequest { + logger.Warn("archive response failed", "retry_after", resp.Header.Get("Retry-After")) + } if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden { resp.Body.Close() return nil, domain.FailIndex(domain.IndexAuthentication, errors.New("archive credentials require operator repair")) @@ -55,7 +80,7 @@ func (t *replayTransport) RoundTrip(req *http.Request) (*http.Response, error) { t.observe(domain.ReplayProgress{Phase: domain.IndexLive}) return resp, nil } - resp.Body = &replayBody{ReadCloser: resp.Body, observe: t.observe, plan: req.URL.Path == "/xrpc/network.bsky.jetstream.planSnapshot" && resp.StatusCode == http.StatusOK} + resp.Body = &replayBody{ReadCloser: resp.Body, observe: t.observe, logger: logger, plan: req.URL.Path == "/xrpc/network.bsky.jetstream.planSnapshot" && resp.StatusCode == http.StatusOK} return resp, nil } @@ -65,25 +90,38 @@ type replayBody struct { plan bool buffer bytes.Buffer finished bool + logger *slog.Logger } func (b *replayBody) Read(p []byte) (int, error) { n, err := b.ReadCloser.Read(p) if n > 0 { b.observe(domain.ReplayProgress{Bytes: int64(n)}) - if b.plan && b.buffer.Len()+n <= MaxEventBytes { - b.buffer.Write(p[:n]) - } else { - b.plan = false + if b.plan { + if b.buffer.Len()+n <= MaxEventBytes { + b.buffer.Write(p[:n]) + } else { + b.plan = false + b.logger.Warn("archive plan observation exceeded buffer limit", "limit_bytes", MaxEventBytes) + } } } if errors.Is(err, io.EOF) && b.plan && !b.finished { b.finished = true var plan struct { - Tip int64 `json:"sealedTipSeq"` + Tip int64 `json:"sealedTipSeq"` + Through int64 `json:"plannedThroughSeq"` + Segments []json.RawMessage `json:"segments"` + Stats struct { + Blocks int64 `json:"blocksMatched"` + Entries int64 `json:"entries"` + } `json:"stats"` } - if json.Unmarshal(b.buffer.Bytes(), &plan) == nil && plan.Tip >= 0 { + if err := json.Unmarshal(b.buffer.Bytes(), &plan); err == nil && plan.Tip >= 0 { b.observe(domain.ReplayProgress{Tip: plan.Tip}) + b.logger.Info("archive plan received", "sealed_tip", plan.Tip, "planned_through", plan.Through, "segments", len(plan.Segments), "blocks", plan.Stats.Blocks, "entries", plan.Stats.Entries) + } else if err != nil { + b.logger.Warn("archive plan observation failed", "error", logging.Error(err)) } } return n, err diff --git a/internal/ui/http.go b/internal/ui/http.go index 4e976bf..e591e27 100644 --- a/internal/ui/http.go +++ b/internal/ui/http.go @@ -5,6 +5,7 @@ import ( "crypto/subtle" _ "embed" "errors" + "fmt" "html/template" "log/slog" "net/http" @@ -311,7 +312,7 @@ func (h *HTTP) fail(w http.ResponseWriter, r *http.Request, err error) { upstreamStatus = api.StatusCode } // Correlate failures without logging tokens, query strings, or remote error bodies. - slog.Warn("UI request failed", "request_id", reference, "method", r.Method, "path", r.URL.Path, "code", problem.code, "upstream_status", upstreamStatus) + slog.Warn("UI request failed", "component", "http", "request_id", reference, "method", r.Method, "path", r.URL.Path, "status", problem.status, "code", problem.code, "upstream_status", upstreamStatus, "error_type", fmt.Sprintf("%T", err)) w.Header().Set("X-Request-ID", reference) if strings.HasPrefix(r.URL.Path, "/oauth/") && !strings.Contains(r.Header.Get("Accept"), "application/json") { w.Header().Set("Content-Type", "text/html; charset=utf-8") diff --git a/internal/upstream/queue.go b/internal/upstream/queue.go index 9ae7c29..8655ce8 100644 --- a/internal/upstream/queue.go +++ b/internal/upstream/queue.go @@ -4,14 +4,18 @@ import ( "context" "errors" "io" + "log/slog" "math/rand/v2" "net" "net/http" "strconv" "strings" "sync" + "sync/atomic" "syscall" "time" + + "provenance/internal/logging" ) const ( @@ -28,10 +32,11 @@ const ( ) type job struct { - req *http.Request - base http.RoundTripper - result chan result - policy RetryPolicy + req *http.Request + base http.RoundTripper + result chan result + policy RetryPolicy + queuedAt time.Time } type result struct { @@ -106,7 +111,7 @@ func (t queued) RoundTrip(req *http.Request) (resp *http.Response, err error) { if err := t.queue.ctx.Err(); err != nil { return nil, err } - work := job{req: req, base: t.base, result: make(chan result), policy: t.policy} + work := job{req: req, base: t.base, result: make(chan result), policy: t.policy, queuedAt: time.Now()} select { case <-req.Context().Done(): return nil, req.Context().Err() @@ -147,10 +152,29 @@ func (q *Queue) serve(work job) { cleanup := func() { stop(); cancel() } req := work.req.Clone(ctx) req.Header.Set("User-Agent", UserAgent) + started := time.Now() + logger := slog.Default().With("component", "external_http", "method", req.Method, "host", req.URL.Host, "path", req.URL.Path, "retry_owner", retryOwner(work.policy)) + logger.Debug("upstream request started", "queue_wait", started.Sub(work.queuedAt)) resp, err := q.fetch(req, work.base, work.policy) + if err != nil { + level := slog.LevelWarn + if errors.Is(err, context.Canceled) { + level = slog.LevelDebug + } + logger.Log(req.Context(), level, "upstream request failed", "duration", time.Since(started), "error", logging.Error(err)) + } var body *heldBody if resp != nil { + logger = logger.With("status", resp.StatusCode) + logger.Debug("upstream response received", "header_duration", time.Since(started), "content_length", resp.ContentLength) + if resp.StatusCode >= http.StatusBadRequest { + logger.Warn("upstream response failed") + } body = &heldBody{ReadCloser: resp.Body, done: make(chan struct{}), cleanup: cleanup, remaining: resp.ContentLength} + body.logger = logger + body.onClose = func() { + logger.Debug("upstream response closed", "duration", time.Since(started), "read_bytes", body.readBytes.Load()) + } if resp.StatusCode == http.StatusSwitchingProtocols { body.remaining = unknownBodyLength } @@ -192,11 +216,24 @@ type heldBody struct { once sync.Once cleanup func() remaining int64 + readBytes atomic.Int64 + logger *slog.Logger + onClose func() } func (b *heldBody) Close() error { var err error - b.once.Do(func() { err = b.ReadCloser.Close(); b.cleanup(); close(b.done) }) + b.once.Do(func() { + err = b.ReadCloser.Close() + b.cleanup() + close(b.done) + if b.onClose != nil { + b.onClose() + } + if err != nil && b.logger != nil { + b.logger.Warn("upstream response close failed", "error", logging.Error(err)) + } + }) return err } @@ -213,6 +250,10 @@ func (b *heldBody) Read(p []byte) (int, error) { return 0, io.EOF } n, err := b.ReadCloser.Read(p) + b.readBytes.Add(int64(n)) + if err != nil && !errors.Is(err, io.EOF) && b.logger != nil { + b.logger.Warn("upstream response read failed", "read_bytes", b.readBytes.Load(), "error", logging.Error(err)) + } // ReadFull can consume a complete range without a final EOF read. Release its worker at the declared length. if b.remaining > 0 { b.remaining -= int64(n) @@ -277,10 +318,22 @@ func (q *Queue) fetch(req *http.Request, base http.RoundTripper, policy RetryPol _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, retryDrainBytes)) resp.Body.Close() } + status := 0 + if resp != nil { + status = resp.StatusCode + } + slog.Warn("upstream retry scheduled", "component", "external_http", "method", req.Method, "host", req.URL.Host, "path", req.URL.Path, "status", status, "attempt", attempt, "next_attempt", attempt+1, "backoff", delay, "error", logging.Error(err)) delay = min(delay*2, retryMax) } } +func retryOwner(policy RetryPolicy) string { + if policy == CallerRetries { + return "caller" + } + return "queue" +} + func transient(status int) bool { return status == http.StatusTooManyRequests || status == http.StatusBadGateway || status == http.StatusServiceUnavailable || status == http.StatusGatewayTimeout || status == http.StatusInternalServerError } @@ -336,6 +389,7 @@ func (q *Queue) cool(ctx context.Context, host string) error { if !until.After(now) { return nil } + slog.Debug("upstream cooldown wait", "component", "external_http", "origin", host, "retry_after", until.Sub(now)) if err := q.wait(ctx, until.Sub(now)); err != nil { return err } diff --git a/internal/upstream/queue_test.go b/internal/upstream/queue_test.go index e2a4056..2d4a70d 100644 --- a/internal/upstream/queue_test.go +++ b/internal/upstream/queue_test.go @@ -1,10 +1,12 @@ package upstream import ( + "bytes" "context" "errors" "fmt" "io" + "log/slog" "net/http" "net/http/httptest" "strings" @@ -13,6 +15,7 @@ import ( "time" "github.com/coder/websocket" + "provenance/internal/logging" ) type transport func(*http.Request) (*http.Response, error) @@ -97,6 +100,43 @@ func TestCompleteReadReleasesWorker(t *testing.T) { } } +func TestRequestLogs(t *testing.T) { + var output bytes.Buffer + previous := slog.Default() + slog.SetDefault(logging.New(&output, slog.LevelDebug, "appview", nil)) + t.Cleanup(func() { slog.SetDefault(previous) }) + const payload = "private-record-content" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte(payload)) })) + defer server.Close() + q, err := New(context.Background(), 1) + if err != nil { + t.Fatal(err) + } + defer q.Close() + client := &http.Client{Transport: q.Transport(server.Client().Transport)} + req, _ := http.NewRequest(http.MethodGet, server.URL+"/download?access_token=query-secret", nil) + req.Header.Set("Authorization", "Bearer header-secret") + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + if _, err := io.ReadAll(resp.Body); err != nil { + t.Fatal(err) + } + resp.Body.Close() + q.Close() + for _, secret := range []string{payload, "query-secret", "header-secret"} { + if strings.Contains(output.String(), secret) { + t.Fatalf("request log exposed %s", secret) + } + } + for _, detail := range []string{`"path":"/download"`, `"status":200`, `"read_bytes":22`, `"queue_wait":`, `"duration":`} { + if !strings.Contains(output.String(), detail) { + t.Errorf("missing request detail: %s", detail) + } + } +} + func TestIncompleteReadHoldsWorker(t *testing.T) { const payload = "range" for _, length := range []int64{int64(len(payload)), unknownBodyLength} { diff --git a/nix/modules/appview.nix b/nix/modules/appview.nix index 17ccca8..3016cfb 100644 --- a/nix/modules/appview.nix +++ b/nix/modules/appview.nix @@ -30,6 +30,7 @@ in stateDirectory = common.stateOption "provenance-appview"; externalWorkers = common.externalWorkersOption; + logLevel = common.logLevelOption; databaseFile = common.databaseOption "appview.db"; serviceKeyFile = common.secretOption "Absolute runtime path to the shared service key. Keep the file outside the Nix store."; @@ -115,6 +116,7 @@ in PROVENANCE_HOP_PENALTY = toString cfg.hopPenalty; PROVENANCE_TRUST_CACHE_TTL = cfg.trustCacheTTL; PROVENANCE_EXTERNAL_WORKERS = toString cfg.externalWorkers; + PROVENANCE_LOG_LEVEL = cfg.logLevel; ENABLE_SOFT_VOUCH = lib.boolToString cfg.enableSoftVouch; ENABLE_SOFT_VOUCH_LISTS = lib.boolToString cfg.softVouchLists; ENABLE_SOFT_VOUCH_ALL_AUTHORS = lib.boolToString cfg.softVouchAllAuthors; diff --git a/nix/modules/common.nix b/nix/modules/common.nix index 51ec9ba..90042ca 100644 --- a/nix/modules/common.nix +++ b/nix/modules/common.nix @@ -11,6 +11,17 @@ in { inherit validSecret; + logLevelOption = lib.mkOption { + type = lib.types.enum [ + "debug" + "info" + "warn" + "error" + ]; + default = "info"; + description = "Minimum structured log level. Debug includes requests and committed replay batches."; + }; + externalWorkersOption = lib.mkOption { type = lib.types.ints.between 1 64; default = 4; diff --git a/nix/modules/ui.nix b/nix/modules/ui.nix index f6411d9..22e74a2 100644 --- a/nix/modules/ui.nix +++ b/nix/modules/ui.nix @@ -31,6 +31,7 @@ in stateDirectory = common.stateOption "provenance-ui"; externalWorkers = common.externalWorkersOption; + logLevel = common.logLevelOption; databaseFile = common.databaseOption "ui.db"; serviceKeyFile = common.secretOption "Absolute runtime path to the appview service key. Keep the file outside the Nix store."; encryptionKeyFile = common.secretOption "Absolute runtime path to the base64-encoded 32-byte encryption key. Keep the file outside the Nix store."; @@ -80,6 +81,7 @@ in PROVENANCE_PUBLIC_ORIGIN = cfg.publicOrigin; PROVENANCE_APPVIEW_URL = cfg.appviewURL; PROVENANCE_EXTERNAL_WORKERS = toString cfg.externalWorkers; + PROVENANCE_LOG_LEVEL = cfg.logLevel; }; }) (lib.mkIf localAppview { after = [ "provenance-appview.service" ]; }) diff --git a/nix/services/provenance.nix b/nix/services/provenance.nix index c514663..d477367 100644 --- a/nix/services/provenance.nix +++ b/nix/services/provenance.nix @@ -72,6 +72,7 @@ let trust_cache_ttl=${lib.escapeShellArg cfg.trustCacheTTL} export PROVENANCE_TRUST_CACHE_TTL="''${PROVENANCE_TRUST_CACHE_TTL:-$trust_cache_ttl}" export PROVENANCE_EXTERNAL_WORKERS="''${PROVENANCE_EXTERNAL_WORKERS:-${toString cfg.externalWorkers}}" + export PROVENANCE_LOG_LEVEL="''${PROVENANCE_LOG_LEVEL:-${cfg.logLevel}}" export PROVENANCE_JETSTREAM_REPLAY="''${PROVENANCE_JETSTREAM_REPLAY:-${lib.boolToString cfg.jetstreamReplay}}" replay_url=${lib.escapeShellArg cfg.jetstreamReplayURL} export PROVENANCE_JETSTREAM_REPLAY_URL="''${PROVENANCE_JETSTREAM_REPLAY_URL:-$replay_url}" @@ -112,6 +113,7 @@ in description = "Trust cache lifetime as a positive Go duration."; }; externalWorkers = (import ../modules/common.nix { inherit lib; }).externalWorkersOption; + logLevel = (import ../modules/common.nix { inherit lib; }).logLevelOption; jetstreamReplay = lib.mkEnableOption "historical Jetstream v2 replay with live cutover"; jetstreamAfterSequence = lib.mkOption { type = lib.types.ints.unsigned; diff --git a/nix/tests/modules.nix b/nix/tests/modules.nix index 8f39139..310b737 100644 --- a/nix/tests/modules.nix +++ b/nix/tests/modules.nix @@ -35,6 +35,7 @@ let hopPenalty = 0.25; trustCacheTTL = "48h"; externalWorkers = 2; + logLevel = "debug"; enableSoftVouch = true; softVouchLists = true; softVouchAllAuthors = true; @@ -124,6 +125,8 @@ assert appUnit.environment.PROVENANCE_SLINGSHOT_URL == "https://records.example. assert appUnit.environment.PROVENANCE_HOP_PENALTY == toString 0.25; assert appUnit.environment.PROVENANCE_TRUST_CACHE_TTL == "48h"; assert appUnit.environment.PROVENANCE_EXTERNAL_WORKERS == "2"; +assert appUnit.environment.PROVENANCE_LOG_LEVEL == "debug"; +assert uiUnit.environment.PROVENANCE_LOG_LEVEL == "info"; assert uiUnit.environment.PROVENANCE_EXTERNAL_WORKERS == "3"; assert appUnit.environment.ENABLE_SOFT_VOUCH == "true"; assert appUnit.environment.ENABLE_SOFT_VOUCH_LISTS == "true"; -- 2.51.2