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";