diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b5590ca..06856e3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -84,10 +84,10 @@ Every change must preserve these project contracts: The `internal/archguard` tests enforce the package and dependency boundaries: - **Core** (`applypatch`, `model`, `redact`, `rule`, `sequence`, `finding`, - `pipeline`, `output`, `winfile`) must not import a plane package. + `pipeline`, `output`, `spool`, `winfile`) must not import a plane package. - **Forensics** (`extract`, `discover`, `casebundle`) and **monitoring** (`hook`, `otel`, `state`) remain independent; neither imports the other. -- Only `state` and `sequence` may import `bbolt`. +- Only `state`, `sequence`, and `spool` may import `bbolt`. - In `cmd/numbat`, scan, timeline, and case files must not import the monitoring plane or bbolt. Hook, collect, and ship files must not import the forensics plane. `agents.go` is the sole cross-plane reporting bridge. diff --git a/README.md b/README.md index b2e0117..c3519f1 100644 --- a/README.md +++ b/README.md @@ -6,8 +6,8 @@ pre-action blocking, and forensic reconstruction. numbat observes supported desktop, CLI, IDE, and gateway agents through local hooks and plugins, OTLP/HTTP logs, and on-disk session artifacts. Live and at-rest activity is normalized into one event model and evaluated by the same -CEL rule engine. Detection runs locally; records can be written to stdout or a -local file and optionally delivered over HTTP. +CEL rule engine. Detection runs locally. Records can use stdout, a local file, +a durable on-disk queue, or HTTP delivery. The [coverage matrix](docs/agent-coverage.md#matrix) is authoritative for each host and surface. Blocking is off by default and limited to supported diff --git a/cmd/numbat/collect.go b/cmd/numbat/collect.go index a9ac311..4a7716c 100644 --- a/cmd/numbat/collect.go +++ b/cmd/numbat/collect.go @@ -65,6 +65,7 @@ func runCollect(args []string, stdout, stderr io.Writer) int { var outputValues multiFlag fs.Var(&outputValues, "output", outputFlagHelp(outputModeStdout)) outputFile := fs.String("output-file", "", "destination path (required when --output includes file)") + spoolFile := fs.String("spool-file", "", "durable queue path (required when --output includes spool)") httpURL := fs.String("http-url", "", "ingest URL (required when --output includes http)") httpBatch := fs.Int("http-batch-size", 500, "records per HTTP POST") httpTimeout := fs.Duration("http-timeout", 30*time.Second, "HTTP request timeout") @@ -129,6 +130,7 @@ func runCollect(args []string, stdout, stderr io.Writer) int { modes: outputValues, defaultMode: outputModeStdout, file: *outputFile, + spool: *spoolFile, httpURL: *httpURL, httpBatch: *httpBatch, httpTimeout: *httpTimeout, diff --git a/cmd/numbat/hook.go b/cmd/numbat/hook.go index 804a4d0..cefb19d 100644 --- a/cmd/numbat/hook.go +++ b/cmd/numbat/hook.go @@ -90,8 +90,9 @@ func runHookEvent(event string, args []string, stdin io.Reader, stdout, stderr i contentFlag := fs.String("content", "preview", contentFlagHelp()) includeReasoning := fs.Bool("include-reasoning", false, "include source-recorded reasoning events when the integration exposes them") var outputValues multiFlag - fs.Var(&outputValues, "output", outputFlagHelp(outputModeStdout)+"; stdout mode writes records to hook stderr and is unavailable in enforce mode") + fs.Var(&outputValues, "output", outputFlagHelp(outputModeStdout)+". stdout mode writes records to hook stderr and is unavailable in enforce mode") outputFile := fs.String("output-file", "", "destination path (required when --output includes file)") + spoolFile := fs.String("spool-file", "", "durable queue path (required when --output includes spool)") httpURL := fs.String("http-url", "", "ingest URL (required when --output includes http)") httpBatch := fs.Int("http-batch-size", 500, "records per HTTP POST") httpTimeout := fs.Duration("http-timeout", defaultHookHTTPTimeout, "HTTP request timeout") @@ -199,6 +200,7 @@ func runHookEvent(event string, args []string, stdin io.Reader, stdout, stderr i stateDB: *stateDB, modes: outputValues, file: *outputFile, + spool: *spoolFile, httpURL: *httpURL, httpBatch: *httpBatch, httpTO: *httpTimeout, @@ -253,6 +255,7 @@ type hookOptions struct { stateDB string modes []string file string + spool string httpURL string httpBatch int httpTO time.Duration @@ -291,10 +294,22 @@ func handleHook(event string, lc hook.Lifecycle, agent, sourceAgent string, stdi if err != nil { return false, "", "", err } + spoolFile, err := expandHookPath(opts.spool) + if err != nil { + return false, "", "", err + } + var statePath string + if sinks.spool { + statePath, err = hookStatePath(opts.stateDB, spoolFile) + if err != nil { + return false, "", "", err + } + } sink, err := buildSink(sinkConfig{ modes: opts.modes, defaultMode: outputModeStdout, file: outputFile, + spool: spoolFile, httpURL: opts.httpURL, httpBatch: opts.httpBatch, httpTimeout: opts.httpTO, @@ -353,7 +368,12 @@ func handleHook(event string, lc hook.Lifecycle, agent, sourceAgent string, stdi var stateErr error if (opts.sel.findings || opts.enforce) && eng != nil { if seqs := eng.SequenceRules(); len(seqs) > 0 { - stateDB, stateErr = openHookState(opts.stateDB) + if statePath == "" { + statePath, stateErr = hookStatePath(opts.stateDB, spoolFile) + } + if stateErr == nil { + stateDB, stateErr = state.Open(statePath, 250*time.Millisecond) + } if stateErr != nil { em.Diag("warn", fmt.Sprintf("state database unavailable, %d sequence rule(s) skipped for this event: %v", len(seqs), stateErr)) } else { @@ -596,7 +616,7 @@ func validateEnforceIO(enforce bool, sel emitSelection, sinks outputSinks) error return fmt.Errorf("--enforce requires --emit findings (or --emit all)") } if sinks.stdout { - return fmt.Errorf("--enforce does not support --output stdout; use file and/or http for operator findings") + return fmt.Errorf("--enforce does not support --output stdout; use file, spool, and/or http for operator findings") } return nil } @@ -687,18 +707,103 @@ func hookEventID(run string) string { // findingOptions pins one detection time across findings from the same callback. func findingOptions() finding.Options { return finding.Options{Now: time.Now()} } -// openHookState opens the shared state database for one hook invocation: the -// --state-db override, or state.db under a numbat-owned directory in the -// user's home. The lock timeout is short because the agent is blocked while -// the hook runs — a contended lock fails open rather than stalling it. -func openHookState(override string) (*state.DB, error) { +// hookStatePath resolves the shared sequence state path and prevents it from +// sharing a bbolt file with the record spool. +func hookStatePath(override, spoolPath string) (string, error) { path := override if path == "" { home, err := os.UserHomeDir() if err != nil { - return nil, fmt.Errorf("resolve home for state db: %w", err) + return "", fmt.Errorf("resolve home for state db: %w", err) } path = filepath.Join(home, ".numbat", "state.db") } - return state.Open(path, 250*time.Millisecond) + if spoolPath != "" { + same, err := sameHookDataPath(spoolPath, path) + if err != nil { + return "", fmt.Errorf("compare --spool-file and --state-db: %w", err) + } + if same { + return "", errors.New("--spool-file and --state-db must name different files") + } + } + return path, nil +} + +func sameHookDataPath(a, b string) (bool, error) { return sameShipPath(a, b) } + +func pathWithResolvedParent(path string) (string, error) { + path, err := filepath.Abs(filepath.Clean(path)) + if err != nil { + return "", err + } + for { + resolved, err := filepath.EvalSymlinks(path) + if err == nil { + return resolved, nil + } + if !errors.Is(err, os.ErrNotExist) { + return "", err + } + info, err := os.Lstat(path) + if errors.Is(err, os.ErrNotExist) || err == nil && info.Mode()&os.ModeSymlink == 0 { + break + } + if err != nil { + return "", err + } + path, err = resolvedSymlinkTarget(path) + if err != nil { + return "", err + } + } + parent := filepath.Dir(path) + var missing []string + for { + resolved, err := filepath.EvalSymlinks(parent) + if err == nil { + for i := len(missing) - 1; i >= 0; i-- { + resolved = filepath.Join(resolved, missing[i]) + } + return filepath.Join(resolved, filepath.Base(path)), nil + } + if !errors.Is(err, os.ErrNotExist) { + return "", err + } + info, lstatErr := os.Lstat(parent) + if lstatErr == nil && info.Mode()&os.ModeSymlink != 0 { + target, err := resolvedSymlinkTarget(parent) + if err != nil { + return "", err + } + for i := len(missing) - 1; i >= 0; i-- { + target = filepath.Join(target, missing[i]) + } + return pathWithResolvedParent(filepath.Join(target, filepath.Base(path))) + } + if lstatErr != nil && !errors.Is(lstatErr, os.ErrNotExist) { + return "", lstatErr + } + next := filepath.Dir(parent) + if next == parent { + return "", err + } + missing = append(missing, filepath.Base(parent)) + parent = next + } +} + +func resolvedSymlinkTarget(path string) (string, error) { + target, err := os.Readlink(path) + if err != nil { + return "", err + } + if !filepath.IsAbs(target) { + parent, err := filepath.EvalSymlinks(filepath.Dir(path)) + if err != nil { + return "", err + } + target = filepath.Join(parent, target) + } + return filepath.Abs(filepath.Clean(target)) } diff --git a/cmd/numbat/hook_install.go b/cmd/numbat/hook_install.go index 3f5a96a..1c7ad68 100644 --- a/cmd/numbat/hook_install.go +++ b/cmd/numbat/hook_install.go @@ -43,6 +43,7 @@ func runHookAdmin(action string, args []string, stdout, stderr io.Writer) int { includeReasoning bool outputValues multiFlag outputFileValue string + spoolFileValue string httpURL string httpBatch = 500 httpTimeout = defaultHookHTTPTimeout @@ -59,8 +60,9 @@ func runHookAdmin(action string, args []string, stdout, stderr io.Writer) int { fs.Var(&emitValues, "emit", "records emitted by live integrations: findings, events, indicators, or all (repeatable; default findings; enforce mode requires findings)") fs.StringVar(&contentValue, "content", "preview", contentFlagHelp()) fs.BoolVar(&includeReasoning, "include-reasoning", false, "include source-recorded reasoning events when an integration exposes them") - fs.Var(&outputValues, "output", outputFlagHelp(outputModeFile)+"; stdout mode writes records to hook stderr and is unavailable in enforce mode") + fs.Var(&outputValues, "output", outputFlagHelp(outputModeFile)+". stdout mode writes records to hook stderr and is unavailable in enforce mode") fs.StringVar(&outputFileValue, "output-file", "", "destination path when --output includes file (default findings.ndjson, or records.ndjson when --emit includes events/indicators)") + fs.StringVar(&spoolFileValue, "spool-file", "", "durable queue path when --output includes spool (default findings.spool, or records.spool when --emit includes events/indicators)") fs.StringVar(&httpURL, "http-url", "", "ingest URL (required when --output includes http)") fs.IntVar(&httpBatch, "http-batch-size", 500, "records per HTTP POST") fs.DurationVar(&httpTimeout, "http-timeout", defaultHookHTTPTimeout, "HTTP request timeout") @@ -189,6 +191,7 @@ func runHookAdmin(action string, args []string, stdout, stderr io.Writer) int { includeReasoning: includeReasoning, modes: outputValues, file: outputFileValue, + spool: spoolFileValue, httpURL: httpURL, httpBatch: httpBatch, httpTimeout: httpTimeout, @@ -289,6 +292,7 @@ type installRuntimeConfig struct { includeReasoning bool modes []string file string + spool string httpURL string httpBatch int httpTimeout time.Duration @@ -377,6 +381,9 @@ func installRuntimeArgs(cfg installRuntimeConfig, home string) ([]string, error) if !sinks.file && cfg.file != "" { return nil, fmt.Errorf("--output-file is only valid when --output includes file") } + if !sinks.spool && cfg.spool != "" { + return nil, fmt.Errorf("--spool-file is only valid when --output includes spool") + } if !sinks.http && cfg.httpURL != "" { return nil, fmt.Errorf("--http-url is only valid when --output includes http") } @@ -388,6 +395,9 @@ func installRuntimeArgs(cfg installRuntimeConfig, home string) ([]string, error) if cfg.httpURL != "" { return nil, fmt.Errorf("--http-url is only valid when --output includes http") } + if cfg.spool != "" { + return nil, fmt.Errorf("--spool-file is only valid when --output includes spool") + } return append(args, "--output=stdout"), nil } @@ -418,6 +428,30 @@ func installRuntimeArgs(cfg installRuntimeConfig, home string) ([]string, error) } args = append(args, "--output-file", file) } + if sinks.spool { + if set["spool-file"] && strings.TrimSpace(cfg.spool) == "" { + return nil, fmt.Errorf("--spool-file must not be empty") + } + path := cfg.spool + if path == "" { + findingsOnly := emitSel.defaultFindingsOnly() + if cfg.managed && findingsOnly { + path = "$HOME/.numbat/findings.spool" + } else if cfg.managed { + path = "$HOME/.numbat/records.spool" + } else if findingsOnly { + path = hook.DefaultFindingsSpoolPath(home) + } else { + path = hook.DefaultRecordsSpoolPath(home) + } + } else if !filepath.IsAbs(path) && !runtimeExpandedPath(path) { + path, err = filepath.Abs(path) + if err != nil { + return nil, fmt.Errorf("resolve --spool-file %q: %w", cfg.spool, err) + } + } + args = append(args, "--spool-file", path) + } if sinks.http { if cfg.httpURL == "" { return nil, fmt.Errorf("--output including http requires --http-url URL") diff --git a/cmd/numbat/main.go b/cmd/numbat/main.go index 5044c8d..194a35a 100644 --- a/cmd/numbat/main.go +++ b/cmd/numbat/main.go @@ -13,7 +13,7 @@ // numbat scan [--agent NAME ... | --path FILE|DIR ...] scan agent artifacts; emit selected NDJSON records // numbat timeline [--agent NAME ... | --path FILE|DIR ...] reconstruct a per-session chronological view // numbat collect [--addr 127.0.0.1:4318] receive OTLP/HTTP protobuf logs; emit records -// numbat ship --input-file F --http-url U tail a local record file to an HTTP endpoint +// numbat ship (--spool-file S | --input-file F) --http-url U send local records to an HTTP endpoint // numbat hook (install|uninstall|status) manage live agent integrations // numbat agents report detected and supported local agents // numbat rules check validate and compile rules; run companion tests @@ -203,7 +203,7 @@ usage: numbat scan [--agent NAME ... | --path FILE|DIR ...] scan agent artifacts; emit records (NDJSON) numbat timeline [--agent NAME ... | --path FILE|DIR ...] reconstruct a per-session chronological view (text|json) numbat collect [--addr 127.0.0.1:4318] receive OTLP/HTTP protobuf logs; emit records - numbat ship --input-file F --http-url U tail a local record file to an HTTP endpoint + numbat ship (--spool-file S | --input-file F) --http-url U send local records to an HTTP endpoint numbat hook EVENT --agent NAME live integration callback (normally not run by hand) numbat hook install --agent NAME|all install numbat's live integrations numbat hook uninstall --agent NAME|all remove numbat-owned live integrations diff --git a/cmd/numbat/scan.go b/cmd/numbat/scan.go index 36c6c63..0b949af 100644 --- a/cmd/numbat/scan.go +++ b/cmd/numbat/scan.go @@ -41,6 +41,7 @@ func runScan(args []string, stdout, stderr io.Writer) int { var outputValues multiFlag fs.Var(&outputValues, "output", outputFlagHelp(outputModeStdout)) outputFile := fs.String("output-file", "", "destination path (required when --output includes file)") + spoolFile := fs.String("spool-file", "", "durable queue path (required when --output includes spool)") httpURL := fs.String("http-url", "", "ingest URL (required when --output includes http)") httpBatch := fs.Int("http-batch-size", 500, "records per HTTP POST") httpTimeout := fs.Duration("http-timeout", 30*time.Second, "HTTP request timeout") @@ -128,6 +129,7 @@ func runScan(args []string, stdout, stderr io.Writer) int { modes: outputValues, defaultMode: outputModeStdout, file: *outputFile, + spool: *spoolFile, httpURL: *httpURL, httpBatch: *httpBatch, httpTimeout: *httpTimeout, diff --git a/cmd/numbat/ship.go b/cmd/numbat/ship.go index b2e41b7..5870138 100644 --- a/cmd/numbat/ship.go +++ b/cmd/numbat/ship.go @@ -15,13 +15,13 @@ import ( "os" "os/signal" "path/filepath" - "runtime" "sort" "strings" "syscall" "time" "github.com/perplexityai/numbat/internal/output" + "github.com/perplexityai/numbat/internal/spool" ) const ( @@ -69,6 +69,7 @@ type shipRead struct { rotated bool reset bool skippedOversized bool + skippedMalformed bool guard []byte } @@ -85,9 +86,10 @@ type shipSinkFactory func() (output.Sink, error) func runShip(args []string, stdout, stderr io.Writer) int { fs := flag.NewFlagSet("ship", flag.ContinueOnError) fs.SetOutput(stderr) - inputPath := fs.String("input-file", "", "append-only NDJSON file to ship (required)") + inputPath := fs.String("input-file", "", "legacy append-only NDJSON file to ship") + spoolPath := fs.String("spool-file", "", "durable record queue to ship (use instead of --input-file)") statePath := fs.String("state-file", "", "delivery checkpoint (default .ship-state)") - poll := fs.Duration("poll", defaultShipPoll, "interval between input-file polls") + poll := fs.Duration("poll", defaultShipPoll, "interval between source polls") httpURL := fs.String("http-url", "", "ingest URL (required)") httpTimeout := fs.Duration("http-timeout", 30*time.Second, "HTTP request timeout") httpAuth := fs.String("http-auth", output.AuthNone, "HTTP delivery auth: none|bearer|hmac-sha256") @@ -96,8 +98,10 @@ func runShip(args []string, stdout, stderr io.Writer) int { httpAllowInsecure := fs.Bool("http-allow-insecure", false, "allow plain http to non-loopback hosts") httpGzip := fs.Bool("http-gzip", false, "gzip the HTTP POST body") fs.Usage = func() { - fmt.Fprintln(stderr, "usage: numbat ship --input-file PATH --http-url URL [--state-file PATH] [--poll DUR] [HTTP options]") - fmt.Fprintln(stderr, "\nTails an append-only numbat NDJSON file to an HTTP endpoint with a durable") + fmt.Fprintln(stderr, "usage: numbat ship (--spool-file PATH | --input-file PATH) --http-url URL [--state-file PATH] [--poll DUR] [HTTP options]") + fmt.Fprintln(stderr, "\nDrains a transactional numbat spool, or tails a legacy append-only NDJSON file,") + fmt.Fprintln(stderr, "to an HTTP endpoint. Spool records are acknowledged only after a successful POST.") + fmt.Fprintln(stderr, "Legacy file input uses a durable") fmt.Fprintln(stderr, "checkpoint. Retained records up to 8 MiB are delivered at least once while the") fmt.Fprintln(stderr, "input and rotated files remain available. Receivers must tolerate duplicates.") fmt.Fprintln(stderr, "Records larger than 8 MiB remain in the input file but are skipped.") @@ -115,8 +119,10 @@ func runShip(args []string, stdout, stderr io.Writer) int { fs.Usage() return 2 } - if strings.TrimSpace(*inputPath) == "" { - fmt.Fprintln(stderr, "ship: --input-file is required") + inputSet := strings.TrimSpace(*inputPath) != "" + spoolSet := strings.TrimSpace(*spoolPath) != "" + if inputSet == spoolSet { + fmt.Fprintln(stderr, "ship: exactly one of --spool-file or --input-file is required") fs.Usage() return 2 } @@ -130,13 +136,29 @@ func runShip(args []string, stdout, stderr io.Writer) int { fs.Usage() return 2 } - if *statePath == "" { - *statePath = *inputPath + ".ship-state" - } - if sameShipPath(*inputPath, *statePath) || sameShipPath(*inputPath, *statePath+".lock") { - fmt.Fprintln(stderr, "ship: --state-file and its lock must differ from --input-file") - fs.Usage() - return 2 + if spoolSet { + if *statePath != "" { + fmt.Fprintln(stderr, "ship: --state-file is only valid with legacy --input-file") + fs.Usage() + return 2 + } + } else { + if *statePath == "" { + *statePath = *inputPath + ".ship-state" + } + for _, candidate := range []string{*statePath, *statePath + ".lock"} { + same, err := sameShipPath(*inputPath, candidate) + if err != nil { + fmt.Fprintf(stderr, "ship: compare input and state paths: %v\n", err) + fs.Usage() + return 2 + } + if same { + fmt.Fprintln(stderr, "ship: --state-file and its lock must differ from --input-file") + fs.Usage() + return 2 + } + } } var httpFlagsSet []string fs.Visit(func(f *flag.Flag) { @@ -171,6 +193,24 @@ func runShip(args []string, stdout, stderr io.Writer) int { _ = s.Close() } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + if spoolSet { + store := spool.New(*spoolPath) + if _, err := store.Peek(maxShipBatchBytes); spoolOpenIsFatal(err) { + fmt.Fprintf(stderr, "ship: open spool: %v\n", err) + return 1 + } + lock, err := acquireShipLock(*spoolPath + ".ship.lock") + if err != nil { + fmt.Fprintf(stderr, "ship: acquire spool shipper lock: %v\n", err) + return 1 + } + defer lock.Close() + fmt.Fprintf(stderr, "numbat ship: shipping spool %s to the configured HTTP endpoint (Ctrl-C to stop)\n", *spoolPath) + return runSpoolShipLoop(ctx, store, *poll, factory, stderr) + } + if err := os.MkdirAll(filepath.Dir(*statePath), 0o700); err != nil { fmt.Fprintf(stderr, "ship: create state directory: %v\n", err) return 1 @@ -182,26 +222,25 @@ func runShip(args []string, stdout, stderr io.Writer) int { } defer lock.Close() - ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) - defer stop() fmt.Fprintf(stderr, "numbat ship: shipping %s to the configured HTTP endpoint (Ctrl-C to stop)\n", *inputPath) return runShipLoop(ctx, *inputPath, *statePath, shipDestinationID(*httpURL), *poll, factory, stderr) } -func runShipLoop(ctx context.Context, inputPath, statePath, destination string, poll time.Duration, factory shipSinkFactory, stderr io.Writer) int { - cursor, err := readShipCursor(statePath, destination) - if err != nil { - fmt.Fprintf(stderr, "ship: read state %s: %v\n", statePath, err) - return 1 - } - if cursor.resetReason != "" { - fmt.Fprintf(stderr, "ship: %s; replaying retained records\n", cursor.resetReason) - cursor.resetReason = "" - } +func spoolOpenIsFatal(err error) bool { + return err != nil && !errors.Is(err, spool.ErrBusy) +} + +func runSpoolShipLoop(ctx context.Context, store spool.Store, poll time.Duration, factory shipSinkFactory, stderr io.Writer) int { + return runShipRetryLoop(ctx, poll, stderr, func() error { + return drainSpoolAvailable(ctx, store, maxShipBatchBytes, factory) + }) +} + +func runShipRetryLoop(ctx context.Context, poll time.Duration, stderr io.Writer, drain func() error) int { stalled := false failures := 0 for { - cursor, err = drainAvailable(ctx, inputPath, statePath, cursor, maxShipBatchBytes, factory, stderr) + err := drain() if ctx.Err() != nil { return 0 } @@ -230,6 +269,49 @@ func runShipLoop(ctx context.Context, inputPath, statePath, destination string, } } +func drainSpoolAvailable(ctx context.Context, store spool.Store, maxBytes int, factory shipSinkFactory) error { + for ctx.Err() == nil { + sent, err := shipSpoolBatch(store, maxBytes, factory) + if err != nil || !sent { + return err + } + } + return nil +} + +func shipSpoolBatch(store spool.Store, maxBytes int, factory shipSinkFactory) (bool, error) { + batch, err := store.Peek(maxBytes) + if err != nil { + return false, fmt.Errorf("read spool: %w", err) + } + if len(batch.Records) == 0 { + return false, nil + } + if err := shipBatch(factory, bytes.Join(batch.Records, nil)); err != nil { + return true, err + } + if err := store.Ack(batch); err != nil { + return true, fmt.Errorf("acknowledge spool batch: %w", err) + } + return true, nil +} + +func runShipLoop(ctx context.Context, inputPath, statePath, destination string, poll time.Duration, factory shipSinkFactory, stderr io.Writer) int { + cursor, err := readShipCursor(statePath, destination) + if err != nil { + fmt.Fprintf(stderr, "ship: read state %s: %v\n", statePath, err) + return 1 + } + if cursor.resetReason != "" { + fmt.Fprintf(stderr, "ship: %s; replaying retained records\n", cursor.resetReason) + cursor.resetReason = "" + } + return runShipRetryLoop(ctx, poll, stderr, func() error { + cursor, err = drainAvailable(ctx, inputPath, statePath, cursor, maxShipBatchBytes, factory, stderr) + return err + }) +} + func shipRetryDelay(base time.Duration, failures int) time.Duration { capDelay := maxShipRetryDelay if base > capDelay { @@ -312,10 +394,12 @@ func drainAvailable(ctx context.Context, inputPath, statePath string, cursor shi if len(batch.blob) == 0 { if batch.n > 0 { // readShipBatch reports n>0 with an empty blob only when it - // skipped an oversized record; the offset must still advance - // past those bytes or the same record blocks the next read. + // skipped an oversized or malformed record; the offset must still + // advance past those bytes or the same record blocks the next read. if batch.skippedOversized { fmt.Fprintf(stderr, "ship: %s: record at offset %d exceeds %d-byte limit; skipped from HTTP delivery and retained in the input file\n", inputPath, cursor.checkpoint.Offset, maxShipRecordBytes) + } else if batch.skippedMalformed { + fmt.Fprintf(stderr, "ship: %s: record at offset %d is not a single JSON object; skipped from HTTP delivery and retained in the input file\n", inputPath, cursor.checkpoint.Offset) } drained := cursor.checkpoint.DrainedFileIDs cursor.checkpoint = newShipCheckpoint( @@ -407,6 +491,7 @@ func readShipBatch(path string, checkpoint shipCheckpoint, maxBytes int64) (ship var buf bytes.Buffer var consumed int64 var skipped bool + var skippedMalformed bool var skippedGuard []byte for buf.Len() == 0 || int64(buf.Len()) < maxBytes { line, readErr := readShipLine(br, maxShipRecordBytes) @@ -423,6 +508,19 @@ func readShipBatch(path string, checkpoint shipCheckpoint, maxBytes int64) (ship break } if line.complete { + if !isShippableRecord(line.bytes) { + // A partial record left by a short append, glued onto the next + // record, is not a single JSON object; ingestion would reject the + // whole batch on it. Ship any good records buffered so far, then + // skip this line on the next empty batch so the queue drains past it. + if buf.Len() > 0 { + break + } + consumed += line.consumed + skippedMalformed = true + skippedGuard = line.bytes + break + } _, _ = buf.Write(line.bytes) consumed += line.consumed } @@ -452,6 +550,7 @@ func readShipBatch(path string, checkpoint shipCheckpoint, maxBytes int64) (ship modTime: info.ModTime(), rotated: source.rotated, skippedOversized: skipped, + skippedMalformed: skippedMalformed, guard: skippedGuard, }, nil } @@ -761,14 +860,92 @@ func shipDestinationID(rawURL string) string { return hex.EncodeToString(sum[:]) } -func sameShipPath(a, b string) bool { - a, errA := filepath.Abs(filepath.Clean(a)) - b, errB := filepath.Abs(filepath.Clean(b)) - if errA != nil || errB != nil { - return false +func sameShipPath(a, b string) (bool, error) { + a, err := pathWithResolvedParent(a) + if err != nil { + return false, fmt.Errorf("resolve %q: %w", a, err) + } + b, err = pathWithResolvedParent(b) + if err != nil { + return false, fmt.Errorf("resolve %q: %w", b, err) + } + if a == b { + return true, nil + } + + aParent, aInfo, err := existingShipPath(a) + if err != nil { + return false, fmt.Errorf("inspect %q: %w", a, err) + } + bParent, bInfo, err := existingShipPath(b) + if err != nil { + return false, fmt.Errorf("inspect %q: %w", b, err) + } + if !os.SameFile(aInfo, bInfo) { + return false, nil } - if runtime.GOOS == "windows" { - return strings.EqualFold(a, b) + aRelative, err := filepath.Rel(aParent, a) + if err != nil { + return false, err + } + bRelative, err := filepath.Rel(bParent, b) + if err != nil { + return false, err + } + if aRelative == "." && bRelative == "." { + return true, nil + } + if aRelative == "." || bRelative == "." { + return false, nil + } + if !filepath.IsLocal(aRelative) || !filepath.IsLocal(bRelative) { + return false, errors.New("resolved path escapes its existing parent") + } + return probeSameShipPath(aParent, aRelative, bRelative) +} + +func existingShipPath(path string) (string, os.FileInfo, error) { + for { + info, err := os.Stat(path) + if err == nil { + return path, info, nil + } + if !errors.Is(err, os.ErrNotExist) { + return "", nil, err + } + parent := filepath.Dir(path) + if parent == path { + return "", nil, err + } + path = parent + } +} + +func probeSameShipPath(parent, aRelative, bRelative string) (same bool, err error) { + root, err := os.MkdirTemp(parent, ".numbat-path-identity-") + if err != nil { + return false, err + } + defer func() { err = errors.Join(err, os.RemoveAll(root)) }() + + aPath := filepath.Join(root, aRelative) + if err := os.MkdirAll(filepath.Dir(aPath), 0o700); err != nil { + return false, err + } + file, err := os.OpenFile(aPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) + if err != nil { + return false, err + } + aInfo, statErr := file.Stat() + if err := errors.Join(statErr, file.Close()); err != nil { + return false, err + } + bInfo, err := os.Stat(filepath.Join(root, bRelative)) + if errors.Is(err, os.ErrNotExist) || errors.Is(err, syscall.ENOTDIR) { + return false, nil + } + if err != nil { + return false, err } - return a == b + return os.SameFile(aInfo, bInfo), nil } diff --git a/cmd/numbat/ship_input.go b/cmd/numbat/ship_input.go index 56d9d55..1dc8169 100644 --- a/cmd/numbat/ship_input.go +++ b/cmd/numbat/ship_input.go @@ -2,6 +2,7 @@ package main import ( "bufio" + "bytes" "crypto/sha256" "encoding/binary" "encoding/hex" @@ -195,6 +196,17 @@ func readShipLine(br *bufio.Reader, maxBytes int) (shipLineRead, error) { } } +// isShippableRecord reports whether line is a single valid JSON object, the only +// shape ingestion accepts. A short append (e.g. a full disk) can leave a record +// with no trailing newline that the next record concatenates onto; the glued +// line is not a single object, and shipping it makes ingestion reject the whole +// batch and stall the queue. spoolSink.Write applies the same check at enqueue +// time so a spooled record can never take that shape. +func isShippableRecord(line []byte) bool { + record := bytes.TrimSpace(line) + return len(record) >= 2 && record[0] == '{' && record[len(record)-1] == '}' && json.Valid(record) +} + func appendShipTail(tail, p []byte) []byte { if len(p) >= shipGuardBytes { return append(tail[:0], p[len(p)-shipGuardBytes:]...) diff --git a/cmd/numbat/ship_platform_unix.go b/cmd/numbat/ship_platform_unix.go index ad52b73..2259074 100644 --- a/cmd/numbat/ship_platform_unix.go +++ b/cmd/numbat/ship_platform_unix.go @@ -50,7 +50,7 @@ func acquireShipLock(path string) (io.Closer, error) { } if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { _ = f.Close() - return nil, fmt.Errorf("another ship process is using this state file: %w", err) + return nil, fmt.Errorf("another ship process is using this source: %w", err) } return &unixShipLock{file: f}, nil } diff --git a/cmd/numbat/ship_platform_windows.go b/cmd/numbat/ship_platform_windows.go index ae6c41c..9ef3098 100644 --- a/cmd/numbat/ship_platform_windows.go +++ b/cmd/numbat/ship_platform_windows.go @@ -41,7 +41,7 @@ func acquireShipLock(path string) (io.Closer, error) { &overlapped, ); err != nil { _ = f.Close() - return nil, fmt.Errorf("another ship process is using this state file: %w", err) + return nil, fmt.Errorf("another ship process is using this source: %w", err) } return &windowsShipLock{file: f}, nil } diff --git a/cmd/numbat/ship_test.go b/cmd/numbat/ship_test.go index 02144c2..b8e4696 100644 --- a/cmd/numbat/ship_test.go +++ b/cmd/numbat/ship_test.go @@ -455,6 +455,48 @@ func TestShipOversizedBatchLineStillShips(t *testing.T) { } } +func TestShipSkipsMalformedLine(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + inputPath := filepath.Join(dir, "records.ndjson") + statePath := filepath.Join(dir, "records.ship-state") + // A short append left record "a" without its newline; the next hook's record + // "b" concatenated onto the same line, so the line is not a single JSON + // object. Record "c" follows on its own line. + glued := `{"record_type":"event","event_id":"a"}{"record_type":"event","event_id":"b"}` + "\n" + good := `{"record_type":"event","event_id":"c"}` + "\n" + appendRaw(t, inputPath, []byte(glued+good)) + + var mu sync.Mutex + var bodies []byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + mu.Lock() + bodies = append(bodies, body...) + mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + cursor, err := drainAvailable(ctx, inputPath, statePath, newTestShipCursor(), maxShipBatchBytes, newSinkFactory(srv.URL), io.Discard) + if err != nil { + t.Fatalf("drain: %v", err) + } + if cursor.checkpoint.Offset != int64(len(glued+good)) { + t.Fatalf("offset=%d/%d: malformed line did not drain", cursor.checkpoint.Offset, len(glued+good)) + } + mu.Lock() + defer mu.Unlock() + if !bytes.Contains(bodies, []byte(`"event_id":"c"`)) { + t.Fatalf("good record after the malformed line was not delivered: %q", bodies) + } + // "event_id":"b" appears only inside the glued line; its delivery would mean + // the malformed line was shipped rather than skipped. + if bytes.Contains(bodies, []byte(`"event_id":"b"`)) { + t.Fatalf("malformed line was delivered: %q", bodies) + } +} + func TestShipBoundsOversizedRecord(t *testing.T) { line, err := readShipLine(bufio.NewReader(strings.NewReader("12345\n")), 4) if err == nil || !strings.Contains(err.Error(), "exceeds") { diff --git a/cmd/numbat/sink.go b/cmd/numbat/sink.go index f2dd983..24fa42c 100644 --- a/cmd/numbat/sink.go +++ b/cmd/numbat/sink.go @@ -1,6 +1,7 @@ package main import ( + "bytes" "fmt" "io" "os" @@ -9,6 +10,7 @@ import ( "github.com/perplexityai/numbat/internal/model" "github.com/perplexityai/numbat/internal/output" + "github.com/perplexityai/numbat/internal/spool" "github.com/perplexityai/numbat/internal/version" ) @@ -16,11 +18,12 @@ import ( const ( outputModeStdout = "stdout" outputModeFile = "file" + outputModeSpool = "spool" outputModeHTTP = "http" ) func outputFlagHelp(defaultMode string) string { - return fmt.Sprintf("where to write records: stdout, file, or http (repeatable; default %s; stdout cannot be combined)", defaultMode) + return fmt.Sprintf("where to write records: stdout, file, spool, or http (repeatable, default %s, stdout cannot be combined)", defaultMode) } // Environment variables carrying HTTP auth secrets. Secrets are never accepted @@ -46,6 +49,7 @@ type sinkConfig struct { modes []string defaultMode string file string + spool string httpURL string httpBatch int httpMaxBuffer int @@ -83,11 +87,10 @@ var httpOnlyFlags = map[string]bool{ // buildSink validates the output flag combination and constructs the records // sink. stdout (the default) wraps the provided writer without taking ownership -// of it. File and HTTP sinks require their respective flags; choosing both fans -// the identical stream to the file and to HTTP. Cross-mode flags (for example -// --output-file without file) are rejected so a mistaken invocation fails loudly -// rather than silently ignoring an argument. HTTP auth secrets are read from the -// environment here, never from a flag. +// of it. File, spool, and HTTP sinks require their respective flags. File and +// spool are mutually exclusive because one path cannot safely carry both +// formats. Cross-mode flags are rejected so a mistaken invocation fails loudly. +// HTTP auth secrets are read from the environment here, never from a flag. func buildSink(cfg sinkConfig, stdout io.Writer) (output.Sink, error) { sel, err := parseOutputSinks(cfg.modes, cfg.defaultMode) if err != nil { @@ -104,6 +107,9 @@ func buildSink(cfg sinkConfig, stdout io.Writer) (output.Sink, error) { if !sel.file && cfg.file != "" { return nil, fmt.Errorf("--output-file is only valid when --output includes file") } + if !sel.spool && cfg.spool != "" { + return nil, fmt.Errorf("--spool-file is only valid when --output includes spool") + } if !sel.http && cfg.httpURL != "" { return nil, fmt.Errorf("--http-url is only valid when --output includes http") } @@ -115,12 +121,18 @@ func buildSink(cfg sinkConfig, stdout io.Writer) (output.Sink, error) { if cfg.httpURL != "" { return nil, fmt.Errorf("--http-url is only valid when --output includes http") } + if cfg.spool != "" { + return nil, fmt.Errorf("--spool-file is only valid when --output includes spool") + } return stdoutSink{stdout}, nil } if sel.file && strings.TrimSpace(cfg.file) == "" { return nil, fmt.Errorf("--output including file requires --output-file PATH") } + if sel.spool && strings.TrimSpace(cfg.spool) == "" { + return nil, fmt.Errorf("--output including spool requires --spool-file PATH") + } if sel.http { if err := validateHTTPAuthFlags(cfg.httpAuth, cfg.httpFlagsSet); err != nil { return nil, err @@ -181,6 +193,9 @@ func buildSink(cfg sinkConfig, stdout io.Writer) (output.Sink, error) { sinks = append(sinks, s) } } + if sel.spool { + sinks = append(sinks, spoolSink{store: spool.New(cfg.spool)}) + } if httpSink != nil { sinks = append(sinks, httpSink) } @@ -202,6 +217,7 @@ func validateHTTPAuthFlags(auth string, flags []string) error { type outputSinks struct { stdout bool file bool + spool bool http bool } @@ -213,6 +229,9 @@ func (s outputSinks) canonicalModes() []string { if s.file { modes = append(modes, outputModeFile) } + if s.spool { + modes = append(modes, outputModeSpool) + } if s.http { modes = append(modes, outputModeHTTP) } @@ -241,6 +260,8 @@ func parseOutputSinks(values []string, defaultMode string) (outputSinks, error) sinks.stdout = true case outputModeFile: sinks.file = true + case outputModeSpool: + sinks.spool = true case outputModeHTTP: sinks.http = true default: @@ -255,13 +276,16 @@ func parseOutputSinks(values []string, defaultMode string) (outputSinks, error) return outputSinks{}, invalidOutputError("") } if sinks.stdout && len(seen) > 1 { - return outputSinks{}, fmt.Errorf("invalid --output %q: stdout cannot be combined with file or http", strings.Join(values, " ")) + return outputSinks{}, fmt.Errorf("invalid --output %q: stdout cannot be combined with file, spool, or http", strings.Join(values, " ")) + } + if sinks.file && sinks.spool { + return outputSinks{}, fmt.Errorf("invalid --output %q: file and spool cannot be combined", strings.Join(values, " ")) } return sinks, nil } func invalidOutputError(raw string) error { - return fmt.Errorf("invalid --output %q: want stdout, file, or http", raw) + return fmt.Errorf("invalid --output %q: want stdout, file, spool, or http", raw) } // httpAuthFromEnv maps the --http-auth mode onto an output.HTTPAuth, reading the @@ -293,3 +317,23 @@ func httpAuthFromEnv(mode string) (output.HTTPAuth, error) { type stdoutSink struct{ io.Writer } func (stdoutSink) Close() error { return nil } + +// spoolSink accepts exactly one complete, supported NDJSON record per Write. +// Put returns only after bbolt commits the whole value, so no repair path is +// needed after an interrupted hook process. +type spoolSink struct{ store spool.Store } + +func (sink spoolSink) Write(p []byte) (int, error) { + if len(p) == 0 || len(p) > maxShipRecordBytes || p[len(p)-1] != '\n' || bytes.Count(p, []byte("\n")) != 1 { + return 0, fmt.Errorf("spool sink: record must be one complete NDJSON line of at most %d bytes", maxShipRecordBytes) + } + if !isShippableRecord(p[:len(p)-1]) { + return 0, fmt.Errorf("spool sink: record must be a JSON object") + } + if err := sink.store.Put(p); err != nil { + return 0, err + } + return len(p), nil +} + +func (spoolSink) Close() error { return nil } diff --git a/cmd/numbat/spool_test.go b/cmd/numbat/spool_test.go new file mode 100644 index 0000000..fd20db9 --- /dev/null +++ b/cmd/numbat/spool_test.go @@ -0,0 +1,195 @@ +package main + +import ( + "bytes" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + + "github.com/perplexityai/numbat/internal/spool" +) + +func TestSpoolSinkDoesNotMergeRecordAfterRejectedPartialWrite(t *testing.T) { + store := spool.New(filepath.Join(t.TempDir(), "records.spool")) + sink := spoolSink{store: store} + first := []byte("{\"n\":1}\n") + partial := []byte("{\"n\":") + second := []byte("{\"n\":2}\n") + + if n, err := sink.Write(first); n != len(first) || err != nil { + t.Fatalf("write first record = (%d, %v), want (%d, nil)", n, err, len(first)) + } + if n, err := sink.Write(partial); n != 0 || err == nil { + t.Fatalf("write partial record = (%d, %v), want (0, error)", n, err) + } + if n, err := sink.Write(second); n != len(second) || err != nil { + t.Fatalf("write second record = (%d, %v), want (%d, nil)", n, err, len(second)) + } + assertQueuedRecords(t, store, first, second) +} + +func TestHookSpoolRejectsStateDatabasePathBeforeWriting(t *testing.T) { + statePath := filepath.Join(t.TempDir(), "state.db") + payload := `{"session_id":"s1","cwd":"/p","tool_name":"Read","tool_input":{"file_path":"/p/file"}}` + stdout, stderr, code := runCLIStdin(payload, + "hook", "pre-tool", "--agent", "claude", "--emit", "events", + "--state-db", statePath, + "--output", "spool", "--spool-file", statePath, + ) + if code != 0 || strings.TrimSpace(stdout) != "{}" { + t.Fatalf("hook must fail open: exit=%d stdout=%q stderr=%q", code, stdout, stderr) + } + if !strings.Contains(stderr, "--spool-file and --state-db must name different files") { + t.Fatalf("stderr = %q, want state/spool collision error", stderr) + } + if _, err := os.Stat(statePath); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("colliding spool unexpectedly created %q (err=%v)", statePath, err) + } +} + +func TestShipSpoolBatchAcknowledgesOnlyDeliveredPrefix(t *testing.T) { + store := spool.New(filepath.Join(t.TempDir(), "records.spool")) + first := []byte("{\"n\":1}\n") + second := []byte("{\"n\":2}\n") + third := []byte("{\"n\":3}\n") + for _, record := range [][]byte{first, second} { + if err := store.Put(record); err != nil { + t.Fatalf("put record: %v", err) + } + } + + requests := make(chan []byte, 2) + var attempts atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + requests <- body + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if attempts.Add(1) == 1 { + http.Error(w, "retry", http.StatusServiceUnavailable) + return + } + if err := store.Put(third); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + wantBody := append(bytes.Clone(first), second...) + if sent, err := shipSpoolBatch(store, maxShipBatchBytes, newSinkFactory(server.URL)); !sent || err == nil { + t.Fatalf("failed delivery = (%v, %v), want (true, error)", sent, err) + } + if got := <-requests; !bytes.Equal(got, wantBody) { + t.Fatalf("failed request body = %q, want %q", got, wantBody) + } + assertQueuedRecords(t, store, first, second) + + if sent, err := shipSpoolBatch(store, maxShipBatchBytes, newSinkFactory(server.URL)); !sent || err != nil { + t.Fatalf("successful delivery = (%v, %v), want (true, nil)", sent, err) + } + if got := <-requests; !bytes.Equal(got, wantBody) { + t.Fatalf("successful request body = %q, want %q", got, wantBody) + } + assertQueuedRecords(t, store, third) +} + +func TestSameShipPathFollowsFilesystemIdentity(t *testing.T) { + root := os.Getenv("NUMBAT_CASE_SENSITIVE_TEST_DIR") + if root == "" { + root = t.TempDir() + } + dir, err := os.MkdirTemp(root, "numbat-path-case-") + if err != nil { + t.Fatalf("create test directory: %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + + lower := filepath.Join(dir, "probe") + upper := filepath.Join(dir, "Probe") + if err := os.WriteFile(lower, []byte("lower"), 0o600); err != nil { + t.Fatalf("write lower-case file: %v", err) + } + _, upperErr := os.Stat(upper) + caseSensitive := os.IsNotExist(upperErr) + if upperErr != nil && !caseSensitive { + t.Fatalf("stat case-variant file: %v", upperErr) + } + lower = filepath.Join(dir, "records.db") + upper = filepath.Join(dir, "Records.db") + same, err := sameShipPath(lower, upper) + if err != nil { + t.Fatalf("compare paths: %v", err) + } + wantSame := !caseSensitive + if same != wantSame { + t.Fatalf("same path = %v, want %v on a case-sensitive=%v filesystem", same, wantSame, caseSensitive) + } + + composedProbe := filepath.Join(dir, "\u00e9-probe") + decomposedProbe := filepath.Join(dir, "e\u0301-probe") + if err := os.WriteFile(composedProbe, []byte("probe"), 0o600); err != nil { + t.Fatalf("write composed Unicode file: %v", err) + } + composedInfo, err := os.Stat(composedProbe) + if err != nil { + t.Fatalf("stat composed Unicode file: %v", err) + } + decomposedInfo, decomposedErr := os.Stat(decomposedProbe) + if decomposedErr != nil && !os.IsNotExist(decomposedErr) { + t.Fatalf("stat decomposed Unicode file: %v", decomposedErr) + } + if decomposedErr == nil && os.SameFile(composedInfo, decomposedInfo) { + composed := filepath.Join(dir, "\u00e9.db") + decomposed := filepath.Join(dir, "e\u0301.db") + same, err := sameShipPath(composed, decomposed) + if err != nil { + t.Fatalf("compare normalization-equivalent paths: %v", err) + } + if !same { + t.Fatal("normalization-equivalent paths compare as distinct") + } + } +} + +func TestSameShipPathResolvesDanglingIntermediateSymlink(t *testing.T) { + root := t.TempDir() + if err := os.Symlink("real", filepath.Join(root, "alias")); err != nil { + t.Skipf("create symlink: %v", err) + } + + aliasPath := filepath.Join(root, "alias", "records.db") + targetPath := filepath.Join(root, "real", "records.db") + same, err := sameShipPath(aliasPath, targetPath) + if err != nil { + t.Fatalf("compare paths through dangling symlink: %v", err) + } + if !same { + t.Fatal("paths that converge through a dangling symlink compare as distinct") + } +} + +func assertQueuedRecords(t *testing.T, store spool.Store, want ...[]byte) { + t.Helper() + batch, err := store.Peek(maxShipBatchBytes) + if err != nil { + t.Fatalf("peek records: %v", err) + } + if len(batch.Records) != len(want) { + t.Fatalf("queued records = %q, want %q", batch.Records, want) + } + for i := range want { + if !bytes.Equal(batch.Records[i], want[i]) { + t.Fatalf("queued record %d = %q, want %q", i, batch.Records[i], want[i]) + } + } +} diff --git a/docs/cli.md b/docs/cli.md index f2f68cd..0a084b4 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -14,7 +14,8 @@ numbat scan [--agent NAME ... | --path FILE|DIR ...] numbat timeline [--agent NAME ... | --path FILE|DIR ...] reconstruct a per-session chronological view numbat collect [--addr HOST:PORT] receive OTLP/HTTP protobuf logs; emit records -numbat ship --input-file F --http-url U tail a local record file to an HTTP endpoint +numbat ship (--spool-file S | --input-file F) --http-url U + send queued or legacy file records to HTTP numbat hook EVENT --agent NAME live integration callback (normally not run by hand) numbat hook install --agent NAME|all install numbat's live integrations numbat hook uninstall --agent NAME|all remove numbat-owned live integrations @@ -102,9 +103,10 @@ agent's transcript. --rules-dir DIR operator rules to add or replace by id (repeatable) --no-builtin-rules load only --rules-dir rules ---output SINK record sink: stdout, file, or http +--output SINK record sink: stdout, file, spool, or http (repeatable; default stdout; stdout cannot be combined) --output-file PATH destination path (required when output includes file) +--spool-file PATH durable queue path (required when output includes spool) ``` HTTP sink flags (used when output includes `http`): @@ -139,6 +141,11 @@ With file output, `scan` creates or truncates the destination and requires its parent directory to exist. `collect` and `hook` append and create missing parent directories. +With spool output, each write commits one complete NDJSON record to a durable +queue. A successful write means that the complete record is committed. Use +`numbat ship --spool-file PATH` to deliver the queue. File and spool output +cannot be combined. Either one can be combined with direct HTTP output. + Repeat `--output` to fan out the same NDJSON stream to more than one sink, for example `--output file --output http`. Direct HTTP is not a durable queue. Its 16 MiB memory buffer rejects any larger record. A failed batch is retried @@ -147,8 +154,8 @@ it is never spooled to disk. After a delivery failure, a full buffer drops the oldest complete records to retain the newest records; the close error reports the drop count. A retry after an ambiguous transport failure can duplicate a batch, so receivers should tolerate repeated record ids. For production -retention, keep file output and let a log forwarder ship it. numbat does not -rotate files itself. +retention, use spool output with `numbat ship`, or use file output with a fleet +forwarder. numbat does not rotate files or manage host storage. ``` # scan only automatically discovered Codex artifacts @@ -271,8 +278,8 @@ A scan stream always terminates with a `scan_summary` whose `status` is `complete`, `partial`, or `error`. `partial` covers parsed runs with artifact failures or record-delivery failures observed before the summary was written. Operational diagnostics normally use a separate NDJSON stream on stderr. After -an enforce-mode hook opens its required file or HTTP sink, diagnostics are -written to that sink so details stay out of the host control channel. +an enforce-mode hook opens its required file, spool, or HTTP sink, diagnostics +are written to that sink so details stay out of the host control channel. Exit codes: `0` when the scan parses at least one artifact and output delivery succeeds; `1` when initialization fails before scanning (home/default-root discovery, `--rules-dir` validation, or rule-engine compilation), when zero @@ -351,9 +358,9 @@ telemetry through the same pipeline `scan` uses. It listens on (the standard OTLP/HTTP path), with identity or gzip content encoding. It serves **OTLP/HTTP only** — there is no gRPC listener (`:4317`), so an agent that defaults to gRPC must be switched to HTTP. -The startup banner, shutdown note, and diagnostics are written to stderr; the -selected record sink (`stdout`, `file`, `http`, or repeated `file` + `http`) -stays reserved for records. +The startup banner, shutdown note, and diagnostics are written to stderr. The +selected record sink stays reserved for records. It can be `stdout`, `file`, +`spool`, `http`, or one durable sink plus `http`. The receiver has no client authentication or TLS. Keep the default loopback bind, or place an off-loopback listener behind network controls and an @@ -378,9 +385,10 @@ IDs so receivers can deduplicate it. or all (repeatable; default findings) --content preview|full conversation content in event output (default preview; full is redacted and bounded to 1 MiB) ---output SINK record sink: stdout, file, or http +--output SINK record sink: stdout, file, spool, or http (repeatable; default stdout; stdout cannot be combined) --output-file PATH destination path (required when output includes file) +--spool-file PATH durable queue path (required when output includes spool) --rules-dir DIR operator rules to add or replace by id (repeatable) --no-builtin-rules load only --rules-dir rules @@ -448,22 +456,30 @@ collector for that telemetry. ## ship -`ship` is an optional forwarder for hosts without an existing log shipper. It -tails a numbat NDJSON file and sends batches to an HTTP endpoint outside the -agent's hook path. Its state advances only after a `2xx`, so eligible retained -records are delivered at-least-once across endpoint outages and process restarts -while the input and its rotations remain available. Records larger than 8 MiB -are not eligible for HTTP delivery, as detailed below. +`ship` sends records to an HTTP endpoint outside the agent's hook path. It can +drain a transactional spool or tail a legacy NDJSON file. + +For spool input, `ship` reads the oldest complete records first. It removes +only the delivered prefix after the endpoint returns `2xx`. Failed delivery +keeps every selected record. A record appended during delivery remains queued +for the next request. + +For legacy file input, the checkpoint advances only after a `2xx`. Eligible +records are delivered at least once while the input and its rotations remain +available. Legacy records larger than 8 MiB, and lines that are not a single +JSON object, are skipped. -Use file-only hook output with `ship`. Selecting direct HTTP on the same hook -would send each record through both paths. +Select exactly one input mode. Use spool-only or file-only hook output with +`ship`. Direct HTTP on the same hook sends each record through both paths. ### ship flags ``` ---input-file PATH append-only NDJSON file to ship (required) ---state-file PATH delivery checkpoint (default .ship-state) ---poll DURATION interval between input-file polls (default 2s) +--spool-file PATH transactional record queue to drain +--input-file PATH legacy append-only NDJSON file to tail + (exactly one input path is required) +--state-file PATH legacy file checkpoint (default .ship-state) +--poll DURATION interval between source polls (default 2s) --http-url URL ingest URL (required) --http-timeout DURATION request timeout (default 30s) --http-auth MODE none, bearer, or hmac-sha256 (default none) @@ -476,27 +492,35 @@ would send each record through both paths. ``` numbat hook install --agent codex --emit all \ - --output file --output-file ~/.numbat/records.ndjson + --output spool --spool-file ~/.numbat/records.spool NUMBAT_HTTP_TOKEN=... numbat ship \ - --input-file ~/.numbat/records.ndjson \ + --spool-file ~/.numbat/records.spool \ --http-url https://ingest.example/numbat \ --http-auth bearer ``` -The default state file is `.ship-state`; override it with -`--state-file`. It binds the acknowledged offset to the input file and endpoint. -On rotation, `ship` drains retained, same-directory record files before moving -to the active file. Keep rotations uncompressed until the state reaches the new -active file; a segment deleted during an outage cannot be recovered. -Changing the endpoint or losing valid state replays retained records. Receivers -must tolerate duplicates, using stable record identifiers where present. +Spool input does not use `--state-file`. The queue stores its own delivery +state. The spool sink rejects partial records, multiple records in one write, +non-object JSON, and records larger than 8 MiB. The queue keeps undelivered +records. numbat does not delete them to free storage. + +For legacy input, the default state file is `.ship-state`. Override +it with `--state-file`. It binds the acknowledged offset to the input file and +endpoint. On rotation, `ship` drains retained files before the active file. +Keep rotations uncompressed until the state reaches the active file. A segment +deleted during an outage cannot be recovered. + +Changing the endpoint or losing valid legacy state replays retained records. +An ambiguous HTTP result can also cause a repeated spool batch. Receivers must +tolerate duplicates and use stable record identifiers where present. -`ship` never truncates or rotates the input. Retention remains the operator's -responsibility, and undelivered records are only as durable as that file and its -host. A complete record larger than 8 MiB remains in the input but is skipped -from HTTP delivery with a stderr diagnostic so later records can continue. -Prefer an existing fleet forwarder when one is already available. +`ship` never truncates or rotates a legacy input. Retention remains the +operator's responsibility. A complete record larger than 8 MiB remains in the +input but is skipped, so later records can continue. A line that is not a single +JSON object, such as two records glued together by an interrupted append, is +skipped the same way so one poisoned line cannot stall the queue. Prefer an +existing fleet forwarder when one is already available. `--http-auth`, `--http-timeout`, `--http-gzip`, the HMAC header options, and `--http-allow-insecure` match the [scan HTTP options](#scan), including the wire @@ -568,10 +592,11 @@ below. (default $HOME/.numbat/state.db) --installed-by NAME provenance marker written by `hook install`; accepted but inert (ignored at runtime) ---output SINK record sink: stdout, file, or http +--output SINK record sink: stdout, file, spool, or http (repeatable; default stdout; stdout cannot be combined and is unavailable in enforce mode) --output-file PATH destination path (required when output includes file) +--spool-file PATH durable queue path (required when output includes spool) --rules-dir DIR operator rules to add or replace by id (repeatable) --no-builtin-rules load only --rules-dir rules @@ -579,15 +604,14 @@ below. On `numbat hook`, stdout is reserved for the agent's control response, so monitor-mode `--output=stdout` records are written to **stderr** instead. -Enforce mode requires `--emit findings` (or `all`) and a `file` and/or `http` -sink; stdout output is rejected so operator findings cannot enter the agent's -control channel. After that sink opens, enforce-mode diagnostics are emitted as -records on it even when only findings were selected; decision failures return -only a generic message on hook stderr. For durable capture use -`--output=file`; repeat `--output file --output http` only when you also want a -direct HTTP delivery attempt. `--emit` has the same record selection as `scan` -and `collect`. Hook HTTP requests default to a five-second timeout so a slow -sink does not consume the agent hook's full execution window. +Enforce mode requires `--emit findings` (or `all`) and a `file`, `spool`, or +`http` sink. Stdout output is rejected, so findings cannot enter the agent's +control channel. After the sink opens, enforce-mode diagnostics are emitted as +records on it. Decision failures return only a generic message on hook stderr. +Use file output with an external forwarder. Use spool output with `numbat ship`. +Add direct HTTP only when you also want an immediate delivery attempt. `--emit` +has the same record selection as `scan` and `collect`. Hook HTTP requests use a +five-second timeout by default. The HTTP sink flags (`--http-url`, `--http-auth`, `--http-batch-size`, `--http-gzip`, `--http-timeout`, `--http-sig-header`, `--http-timestamp-header`, @@ -630,13 +654,16 @@ default. This agent process deadline is separate from the hook handler's --include-reasoning include source-recorded reasoning events when the integration exposes them --output SINK record sink installed hook commands use: - stdout, file, or http (repeatable; default file; + stdout, file, spool, or http (repeatable; default file; stdout cannot be combined and writes records to hook stderr because hook stdout is reserved for the agent response; unavailable in enforce mode) --output-file PATH destination path when output includes file (default findings.ndjson for findings only; records.ndjson when events/indicators are selected) +--spool-file PATH queue path when output includes spool + (default findings.spool for findings only; + records.spool when events/indicators are selected) --rules-dir DIR operator rules installed hooks add or replace by id (repeatable) --no-builtin-rules install hook commands that load only --rules-dir @@ -657,14 +684,12 @@ Each `--rules-dir` must be a concrete, readable install-time path; deferred paths such as `$HOME/rules` are rejected in this mode. Install-time output flags are baked into the command written to the agent's hook -configuration. Findings-only installs write -`$HOME/.numbat/findings.ndjson`; selecting events or indicators changes the -default to `$HOME/.numbat/records.ndjson`. Use `--output-file PATH` to change -that file, or -`--output file --output http --http-url URL` to keep the local file and also -attempt direct delivery. HTTP auth secrets are not written into hook settings; the -installed hook reads `NUMBAT_HTTP_TOKEN` or `NUMBAT_HTTP_HMAC_KEY` from its -runtime environment when `--http-auth` selects one of those modes. +configuration. File output uses `$HOME/.numbat/findings.ndjson` for findings +only. It uses `$HOME/.numbat/records.ndjson` when events or indicators are +selected. Spool output uses the corresponding `.spool` names. Use +`--output-file PATH` or `--spool-file PATH` to change the selected destination. +HTTP auth secrets are not written into hook settings. The installed hook reads +`NUMBAT_HTTP_TOKEN` or `NUMBAT_HTTP_HMAC_KEY` from its runtime environment. `hook install` accepts the same eight HTTP flags listed under [hook](#hook-flags), with the same defaults, including the five-second diff --git a/docs/deployment.md b/docs/deployment.md index 72aa3af..4c26a18 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -18,8 +18,8 @@ numbat supports macOS, Linux, and native Windows endpoints. One release binary provides every mode. `hook install` writes that binary's absolute path into agent-specific integration artifacts; each lifecycle event launches a short-lived `numbat hook ...` callback. `ship` is an optional, -separate long-lived process of the same binary that tails the callback's file -output. It is not a second binary, daemon dependency, or hook receiver. +separate long-lived process of the same binary. It drains a transactional spool +or tails legacy file output. It is not a second binary or hook receiver. ## Local use @@ -57,24 +57,36 @@ numbat hook install --agent claude \ --http-url https://ingest.example/numbat ``` -Direct HTTP is not a disk queue. If the endpoint is down, numbat reports delivery -failure for that run; it does not spool hours of records for later replay. Keep -file output enabled and let your existing log forwarder, EDR, or OS retention -policy ship and rotate the file. In `collect` mode, an OTLP success response -confirms local acceptance, not downstream acknowledgement by the HTTP sink. +Direct HTTP is not a disk queue. If the endpoint is down, numbat reports a +delivery failure for that run. Keep file output for an external forwarder. Use +spool output with `numbat ship` when the host has no external forwarder. In +`collect` mode, an OTLP success response confirms local acceptance. It does not +confirm downstream delivery by the HTTP sink. `collect` has no client authentication or TLS. Keep its default loopback bind, or place an off-loopback listener behind network controls and an authenticated proxy. -Where the host has no external shipper, `numbat ship` is an optional native -forwarder that tails the file output and delivers eligible retained records -at-least-once while their input segments remain available, off the hook's -critical path (see [cli.md](cli.md#ship)). Records larger than 8 MiB are skipped. -It uses the capture file as its only on-disk queue and does not replace a mature -shipper where one already runs. Configure -the hook with file output only; combining direct HTTP with `ship` sends the same -records through both paths. +If the host has no external forwarder, configure a transactional spool: + +```bash +numbat hook install --agent codex --emit all \ + --output spool \ + --spool-file ~/.numbat/live.spool + +numbat ship \ + --spool-file ~/.numbat/live.spool \ + --http-url https://ingest.example/numbat +``` + +The hook commits complete records without waiting for the network. `ship` +removes records only after successful HTTP delivery. An endpoint outage keeps +the undelivered records. numbat does not delete those records to manage storage. +Supervise the `ship` process and monitor the spool filesystem. + +`ship --input-file` remains available for legacy file output. Prefer an +existing fleet forwarder when one is already available. Do not combine direct +HTTP with a later `ship` path unless the receiver expects a second copy. ## Choose an install scope diff --git a/docs/enforcement.md b/docs/enforcement.md index 680c25e..07d2f99 100644 --- a/docs/enforcement.md +++ b/docs/enforcement.md @@ -140,9 +140,9 @@ deny_message: Contact the workspace owner. Use an override only when that text is safe to disclose. The host decides whether its reason is shown to the model, the user, or both. -Operator detail travels separately. Enforce mode requires findings to be -selected and written to a `file` and/or `http` sink; stdout is rejected because -hook stdout is part of the host control protocol. Once the sink is open, +Operator detail travels separately. Enforce mode requires findings and a +`file`, `spool`, or `http` sink. It rejects stdout because hook stdout is part +of the host control protocol. Once the sink is open, numbat routes available operational diagnostics there as diagnostic records rather than onto the immediate agent channel. diff --git a/docs/live-capture.md b/docs/live-capture.md index 1d27f2e..c69881d 100644 --- a/docs/live-capture.md +++ b/docs/live-capture.md @@ -116,20 +116,26 @@ bounded previews by default; add exposes it. Add `--include-reasoning` for source-exposed Pi, OpenCode, and Kilo reasoning; hidden model chain-of-thought is not reconstructed. Repeat `--output file --output http --http-url URL` to keep the file and also attempt -direct HTTP delivery. Direct HTTP alone is not durable: it has bounded in-memory -buffering, but no disk spool. numbat does not rotate output files itself; for -long-running all-event streams, let your log forwarder or OS retention policy -manage the file. Hook HTTP requests use a five-second timeout by default; change -it with `--http-timeout`. Agents normally wait for the callback process to exit, -so direct HTTP adds request latency on the hook path. HTTP auth secrets are never -written into hook settings; -when `--http-auth` is `bearer` or `hmac-sha256`, the installed hook reads +direct HTTP delivery. Direct HTTP alone is not durable. Select `--output spool` +for a transactional disk queue, then run `numbat ship --spool-file PATH`. +Spool output defaults to `$HOME/.numbat/findings.spool` or +`$HOME/.numbat/records.spool`. Change it with `--spool-file PATH`. + +numbat does not rotate output files or manage host storage. Use a fleet +forwarder for file output. A short append (for example, a full disk) is rolled +back and a missing trailing newline is repaired before the next write, so +records never concatenate onto a partial line. Supervise `numbat ship` for +spool output. Hook HTTP requests use a five-second timeout by default; change it +with `--http-timeout`. +Agents normally wait for the callback process to exit, so direct HTTP adds +request latency on the hook path. HTTP auth secrets are never written into hook +settings. When `--http-auth` is `bearer` or `hmac-sha256`, the installed hook reads `NUMBAT_HTTP_TOKEN` or `NUMBAT_HTTP_HMAC_KEY` from the agent's runtime environment. The live hook handler always reserves stdout for the agent's allow/deny response (zero bytes on successful Kiro hooks), so `--output=stdout` writes records to stderr in hook mode and is mainly for manual monitor-mode testing. Enforce mode requires findings -and an out-of-band `file` and/or `http` sink; it rejects stdout output so +and an out-of-band `file`, `spool`, or `http` sink. It rejects stdout output, so finding details cannot enter the immediate agent control response. The deployer is responsible for restricting the agent's filesystem or network access to that sink when stronger isolation is required. @@ -141,14 +147,15 @@ Gemini's millisecond-based hooks use five seconds; Junie's `UserPromptSubmit` uses 10 seconds and `SessionEnd` uses two seconds. This bounds a stuck callback instead of inheriting Claude Code or Codex's ten-minute default. It is a process deadline imposed by the agent, distinct from -`--http-timeout`, which bounds one HTTP request inside numbat. Prefer file output -plus a forwarder when delivery may take longer than the hook's deadline. +`--http-timeout`, which bounds one HTTP request inside numbat. Use file output +with an external forwarder, or use spool output with `numbat ship`. Examples: ``` numbat hook install --agent codex --emit all --output-file ~/.numbat/codex.ndjson numbat hook install --agent claude --output file --output http --http-url https://ingest.example/numbat +numbat hook install --agent codex --emit all --output spool --spool-file ~/.numbat/codex.spool ``` PowerShell uses the same flags: @@ -221,13 +228,16 @@ metrics, so trace-only exporters such as VS Code Copilot Chat and OpenHands observability should use hooks or a full OpenTelemetry collector. The `/v1/logs` endpoint and per-agent setup are documented in [cli.md](cli.md#collect). -## Delivering files off-host +## Delivering records off-host With file-only output, hooks do not wait on the network. The file is the durable record stream; ship it with the fleet's existing log forwarder, EDR, or OS -retention tooling. Where the host has no such shipper, `numbat ship` is an -optional native forwarder that tails that file and -delivers eligible retained records at-least-once to an HTTP endpoint while their -input segments remain available; records larger than 8 MiB are skipped. See -[cli.md](cli.md#ship) for the complete limits. Use file-only hook output with -`ship` so the same record is not also sent through direct HTTP. +retention tooling. + +If the host has no external forwarder, select spool-only output. Run +`numbat ship --spool-file PATH` as a supervised process. Failed HTTP delivery +keeps the queued records. Successful delivery removes only the delivered +prefix. See [cli.md](cli.md#ship) for the complete contract. + +`numbat ship` also accepts a legacy file through `--input-file`. Use one local +durable output with `ship`. Direct HTTP on the same hook sends a second copy. diff --git a/internal/archguard/archguard_test.go b/internal/archguard/archguard_test.go index 164ae5c..a1ec104 100644 --- a/internal/archguard/archguard_test.go +++ b/internal/archguard/archguard_test.go @@ -29,7 +29,7 @@ const internalPrefix = modulePath + "/internal/" // Plane membership. Names are the package's path under internal/ (e.g. "model" // or a nested package such as "rule/foo"). var ( - corePkgs = []string{"applypatch", "model", "redact", "rule", "sequence", "finding", "pipeline", "output", "winfile"} + corePkgs = []string{"applypatch", "model", "redact", "rule", "sequence", "finding", "pipeline", "output", "spool", "winfile"} forensicsPkgs = []string{"extract", "discover", "casebundle"} monitoringPkgs = []string{"hook", "otel", "state"} ) @@ -41,7 +41,7 @@ var ( var nonPlane = map[string]bool{"version": true, "archguard": true} // bboltAllowed lists the only internal packages permitted to import bbolt. -var bboltAllowed = map[string]bool{"state": true, "sequence": true} +var bboltAllowed = map[string]bool{"state": true, "sequence": true, "spool": true} // --------------------------------------------------------------------------- // A2a — package-level guard (go list) @@ -117,9 +117,9 @@ func TestPackageImportSeam(t *testing.T) { if monitoring[pkg] && inPlane(forensics, imp) { t.Errorf("monitoring package %q imports %q (monitoring must stay independent of the forensics plane)", pkg, imp) } - // bbolt allowlist: only {state, sequence} may touch it. + // bbolt allowlist: only state, sequence, and spool may touch it. if imp == bboltPath && !bboltAllowed[pkg] { - t.Errorf("package %q imports %s but is not on the bbolt allowlist {state, sequence}", pkg, bboltPath) + t.Errorf("package %q imports %s but is not on the bbolt allowlist {state, sequence, spool}", pkg, bboltPath) } } } diff --git a/internal/hook/install.go b/internal/hook/install.go index 42c6ab9..5f4777d 100644 --- a/internal/hook/install.go +++ b/internal/hook/install.go @@ -722,6 +722,18 @@ func DefaultRecordsPath(home string) string { return filepath.Join(home, ".numbat", "records.ndjson") } +// DefaultFindingsSpoolPath is the transactional queue used by hook installs +// that select spool output and emit only findings. +func DefaultFindingsSpoolPath(home string) string { + return filepath.Join(home, ".numbat", "findings.spool") +} + +// DefaultRecordsSpoolPath is used when a spool hook also emits events or +// indicators. +func DefaultRecordsSpoolPath(home string) string { + return filepath.Join(home, ".numbat", "records.spool") +} + // InstallOptions are the runtime flags baked into commands written by Install. // RuntimeArgs are appended to every installed `numbat hook ` invocation // after the agent marker. Claude receives an argument array; string-only hook diff --git a/internal/output/filesink.go b/internal/output/filesink.go index a25038a..0b16fea 100644 --- a/internal/output/filesink.go +++ b/internal/output/filesink.go @@ -1,7 +1,9 @@ package output import ( + "errors" "fmt" + "io" "os" "path/filepath" ) @@ -69,19 +71,57 @@ func openFileSink(path string, appendMode bool) (Sink, error) { f.Close() return nil, fmt.Errorf("file sink: tighten perms on %q: %w", path, err) } - return &fileSink{file: f}, nil + return &fileSink{file: f, appendMode: appendMode}, nil } // fileSink serializes each write at the file descriptor so separate hook // processes appending to the same records file cannot interleave NDJSON lines. type fileSink struct { - file *os.File + file *os.File + appendMode bool } func (s *fileSink) Write(p []byte) (int, error) { - return writeFileLocked(s.file, p) + return writeFileLocked(s.file, p, s.appendMode) } func (s *fileSink) Close() error { return s.file.Close() } + +// appendRecordLocked writes one complete record to f while the caller holds the +// file lock. It keeps the file from ever ending mid-record so the next record +// cannot concatenate onto a partial line: a missing trailing newline left by an +// earlier short append is repaired first (append sinks only, which open for +// reading), and a short or failed write is rolled back to the pre-write size. +func appendRecordLocked(f *os.File, p []byte, repairNewline bool) (int, error) { + info, err := f.Stat() + if err != nil { + return 0, err + } + size := info.Size() + if repairNewline && size > 0 { + last := make([]byte, 1) + if _, err := f.ReadAt(last, size-1); err != nil { + return 0, err + } + if last[0] != '\n' { + if _, err := f.Write([]byte{'\n'}); err != nil { + _ = f.Truncate(size) + return 0, err + } + size++ + } + } + n, err := f.Write(p) + if err == nil && n == len(p) { + return n, nil + } + if truncErr := f.Truncate(size); truncErr != nil { + err = errors.Join(err, truncErr) + } + if err == nil { + err = io.ErrShortWrite + } + return n, err +} diff --git a/internal/output/filesink_diskfull_unix_test.go b/internal/output/filesink_diskfull_unix_test.go new file mode 100644 index 0000000..b0779d8 --- /dev/null +++ b/internal/output/filesink_diskfull_unix_test.go @@ -0,0 +1,123 @@ +//go:build unix + +package output + +import ( + "bytes" + "fmt" + "os" + "os/exec" + "os/signal" + "path/filepath" + "strconv" + "syscall" + "testing" +) + +const ( + fileSinkDiskFullHelper = "NUMBAT_FILESINK_DISKFULL_HELPER" + fileSinkDiskFullPath = "NUMBAT_FILESINK_DISKFULL_PATH" + fileSinkDiskFullLimit = "NUMBAT_FILESINK_DISKFULL_LIMIT" + fileSinkSeedRecord = `{"record_type":"event","event_id":"seed"}` + "\n" + fileSinkNextRecord = `{"record_type":"event","event_id":"next"}` + "\n" +) + +// TestFileSinkShortAppendRollsBack reproduces the failure that corrupted the +// legacy records file: an append that could not grow the file left a partial +// NDJSON line, and the next hook's record concatenated onto it. A short write +// must now roll back to the pre-write size so the file ends on a record boundary +// and the next record starts on its own line. +// +// RLIMIT_FSIZE, set a few bytes past the seed record in a child process (so the +// cap cannot disturb the test harness), forces the next append to write a +// fragment and then fail without a real full filesystem. +func TestFileSinkShortAppendRollsBack(t *testing.T) { + if os.Getenv(fileSinkDiskFullHelper) == "1" { + runFileSinkShortAppendHelper() + return + } + + path := filepath.Join(t.TempDir(), "records.ndjson") + seed, err := NewFileSinkAppend(path) + if err != nil { + t.Fatalf("open seed sink: %v", err) + } + if _, err := seed.Write([]byte(fileSinkSeedRecord)); err != nil { + t.Fatalf("seed write: %v", err) + } + if err := seed.Close(); err != nil { + t.Fatalf("close seed sink: %v", err) + } + + cmd := exec.Command(os.Args[0], "-test.run=^TestFileSinkShortAppendRollsBack$") + cmd.Env = append(os.Environ(), + fileSinkDiskFullHelper+"=1", + fileSinkDiskFullPath+"="+path, + fileSinkDiskFullLimit+"="+strconv.Itoa(len(fileSinkSeedRecord)+8), + ) + out, err := cmd.CombinedOutput() + if err == nil { + t.Fatalf("child append unexpectedly succeeded under a file-size cap: %s", out) + } + if !bytes.Contains(out, []byte("append-failed")) { + t.Fatalf("child did not report a failed append: %s", out) + } + + // The failed append kept none of its bytes: the file still holds exactly the + // seed record with no trailing fragment. + if b, err := os.ReadFile(path); err != nil { + t.Fatal(err) + } else if string(b) != fileSinkSeedRecord { + t.Fatalf("file after short append = %q, want the seed record only", b) + } + + // The next record appends cleanly instead of gluing onto a partial line. + next, err := NewFileSinkAppend(path) + if err != nil { + t.Fatalf("open next sink: %v", err) + } + if _, err := next.Write([]byte(fileSinkNextRecord)); err != nil { + t.Fatalf("next write: %v", err) + } + if err := next.Close(); err != nil { + t.Fatalf("close next sink: %v", err) + } + if b, err := os.ReadFile(path); err != nil { + t.Fatal(err) + } else if string(b) != fileSinkSeedRecord+fileSinkNextRecord { + t.Fatalf("file = %q, want the seed then next record", b) + } +} + +func runFileSinkShortAppendHelper() { + // Exceeding RLIMIT_FSIZE raises SIGXFSZ, whose default action kills the + // process; ignore it so the offending write returns a short count instead. + signal.Ignore(syscall.SIGXFSZ) + limit, err := strconv.ParseInt(os.Getenv(fileSinkDiskFullLimit), 10, 64) + if err != nil { + reportFileSinkHelper("bad-limit:", err) + } + rlimit := syscall.Rlimit{Cur: uint64(limit), Max: uint64(limit)} + if err := syscall.Setrlimit(syscall.RLIMIT_FSIZE, &rlimit); err != nil { + reportFileSinkHelper("setrlimit-failed:", err) + } + sink, err := NewFileSinkAppend(os.Getenv(fileSinkDiskFullPath)) + if err != nil { + reportFileSinkHelper("open-failed:", err) + } + record := append([]byte(`{"record_type":"event","event_id":"big","payload":"`), bytes.Repeat([]byte("a"), 1<<20)...) + record = append(record, []byte("\"}\n")...) + if _, err := sink.Write(record); err != nil { + reportFileSinkHelper("append-failed:", err) + } + reportFileSinkHelper("append-succeeded", nil) +} + +func reportFileSinkHelper(marker string, cause error) { + if cause != nil { + _, _ = fmt.Fprintln(os.Stdout, marker, cause) + os.Exit(1) + } + _, _ = fmt.Fprintln(os.Stdout, marker) + os.Exit(0) +} diff --git a/internal/output/filesink_nofollow.go b/internal/output/filesink_nofollow.go index 01c2773..3502931 100644 --- a/internal/output/filesink_nofollow.go +++ b/internal/output/filesink_nofollow.go @@ -8,19 +8,20 @@ import ( "syscall" ) -// openNoFollow opens path for create/write-only with O_NOFOLLOW so a symlink at -// the final path component is refused (the kernel returns ELOOP) rather than -// followed. This stops numbat being redirected to truncate/write an arbitrary -// target via a planted symlink at an attacker-influenced output path. When -// appendMode is false the file is truncated (scan's fresh-file-per-run -// behavior); when true it is opened for append (the hook handler accumulates -// findings across repeated per-event invocations into one durable file). +// openNoFollow opens path for create with O_NOFOLLOW so a symlink at the final +// path component is refused (the kernel returns ELOOP) rather than followed. +// This stops numbat being redirected to truncate/write an arbitrary target via a +// planted symlink at an attacker-influenced output path. When appendMode is +// false the file is truncated write-only (scan's fresh-file-per-run behavior); +// when true it is opened read-write for append (the hook handler accumulates +// findings across repeated per-event invocations into one durable file, and read +// access lets writeFileLocked repair a missing trailing newline). func openNoFollow(path string, perm os.FileMode, appendMode bool) (*os.File, error) { - flags := os.O_CREATE | os.O_WRONLY | syscall.O_NOFOLLOW + flags := os.O_CREATE | syscall.O_NOFOLLOW if appendMode { - flags |= os.O_APPEND + flags |= os.O_RDWR | os.O_APPEND } else { - flags |= os.O_TRUNC + flags |= os.O_WRONLY | os.O_TRUNC } return os.OpenFile(path, flags, perm) } @@ -32,12 +33,12 @@ func isNoFollowErr(err error) bool { return errors.Is(err, syscall.ELOOP) } -func writeFileLocked(f *os.File, p []byte) (int, error) { +func writeFileLocked(f *os.File, p []byte, repairNewline bool) (int, error) { if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil { return 0, err } defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }() - return f.Write(p) + return appendRecordLocked(f, p, repairNewline) } diff --git a/internal/output/filesink_other.go b/internal/output/filesink_other.go index 1a156a9..5844e9f 100644 --- a/internal/output/filesink_other.go +++ b/internal/output/filesink_other.go @@ -7,13 +7,14 @@ import "os" // openNoFollow falls back to a plain create open on platforms without // O_NOFOLLOW. The symlink hardening is a no-op there; the fchmod tightening in // NewFileSink still applies. appendMode selects append vs truncate, matching the -// unix build. +// unix build, and opens read-write for append so writeFileLocked can repair a +// missing trailing newline. func openNoFollow(path string, perm os.FileMode, appendMode bool) (*os.File, error) { - flags := os.O_CREATE | os.O_WRONLY + flags := os.O_CREATE if appendMode { - flags |= os.O_APPEND + flags |= os.O_RDWR | os.O_APPEND } else { - flags |= os.O_TRUNC + flags |= os.O_WRONLY | os.O_TRUNC } return os.OpenFile(path, flags, perm) } @@ -21,6 +22,6 @@ func openNoFollow(path string, perm os.FileMode, appendMode bool) (*os.File, err // isNoFollowErr is always false where O_NOFOLLOW is unavailable. func isNoFollowErr(error) bool { return false } -func writeFileLocked(f *os.File, p []byte) (int, error) { - return f.Write(p) +func writeFileLocked(f *os.File, p []byte, repairNewline bool) (int, error) { + return appendRecordLocked(f, p, repairNewline) } diff --git a/internal/output/filesink_test.go b/internal/output/filesink_test.go index bd7378b..858c8e4 100644 --- a/internal/output/filesink_test.go +++ b/internal/output/filesink_test.go @@ -184,6 +184,35 @@ func TestFileSinkAppendSerializesConcurrentWriters(t *testing.T) { } } +// A records file left without a trailing newline by an older numbat's short +// append must be closed off before the next record, so the two never land on the +// same line and ingestion does not reject the glued result. +func TestFileSinkAppendRepairsMissingNewline(t *testing.T) { + path := filepath.Join(t.TempDir(), "records.ndjson") + fragment := `{"record_type":"event","event_id":"frag"` + if err := os.WriteFile(path, []byte(fragment), 0o600); err != nil { + t.Fatal(err) + } + sink, err := NewFileSinkAppend(path) + if err != nil { + t.Fatal(err) + } + record := `{"record_type":"event","event_id":"whole"}` + "\n" + if _, err := sink.Write([]byte(record)); err != nil { + t.Fatalf("append: %v", err) + } + if err := sink.Close(); err != nil { + t.Fatal(err) + } + b, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if want := fragment + "\n" + record; string(b) != want { + t.Fatalf("file = %q, want the fragment and record on separate lines", b) + } +} + func TestFileSinkTruncatesExisting(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "out.ndjson") diff --git a/internal/output/filesink_windows.go b/internal/output/filesink_windows.go index f88a0d4..cf39bd7 100644 --- a/internal/output/filesink_windows.go +++ b/internal/output/filesink_windows.go @@ -17,12 +17,12 @@ func isNoFollowErr(err error) bool { return errors.Is(err, winfile.ErrReparsePoint) } -func writeFileLocked(f *os.File, p []byte) (n int, err error) { +func writeFileLocked(f *os.File, p []byte, repairNewline bool) (n int, err error) { if err := winfile.LockExclusive(f); err != nil { return 0, err } defer func() { err = errors.Join(err, winfile.Unlock(f)) }() - return f.Write(p) + return appendRecordLocked(f, p, repairNewline) } diff --git a/internal/spool/open_other.go b/internal/spool/open_other.go new file mode 100644 index 0000000..0b480c4 --- /dev/null +++ b/internal/spool/open_other.go @@ -0,0 +1,26 @@ +//go:build !unix && !windows + +package spool + +import ( + "errors" + "os" +) + +func validateParentMode(os.FileInfo) error { return nil } + +func openExistingDatabaseFile(path string, flag int) (*os.File, error) { + return os.OpenFile(path, flag, 0) +} + +func validateDatabaseMode(os.FileInfo) error { return nil } + +func installDatabaseFile(candidate, path string) error { + if err := os.Link(candidate, path); err != nil { + if errors.Is(err, os.ErrExist) { + return nil + } + return err + } + return nil +} diff --git a/internal/spool/open_unix.go b/internal/spool/open_unix.go new file mode 100644 index 0000000..4f37acd --- /dev/null +++ b/internal/spool/open_unix.go @@ -0,0 +1,43 @@ +//go:build unix + +package spool + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "syscall" +) + +func validateParentMode(info os.FileInfo) error { + if info.Mode().Perm()&0o022 != 0 { + return errors.New("directory must not be group- or other-writable") + } + return nil +} + +func openExistingDatabaseFile(path string, flag int) (*os.File, error) { + return os.OpenFile(path, flag|syscall.O_NOFOLLOW, 0) +} + +func validateDatabaseMode(info os.FileInfo) error { + if info.Mode().Perm()&0o077 != 0 { + return fmt.Errorf("existing database permissions are %04o, want 0600 or stricter", info.Mode().Perm()) + } + return nil +} + +func installDatabaseFile(candidate, path string) error { + if err := os.Link(candidate, path); err != nil { + if errors.Is(err, os.ErrExist) { + return nil + } + return err + } + dir, err := os.Open(filepath.Dir(path)) + if err != nil { + return err + } + return errors.Join(dir.Sync(), dir.Close()) +} diff --git a/internal/spool/open_windows.go b/internal/spool/open_windows.go new file mode 100644 index 0000000..ec5b145 --- /dev/null +++ b/internal/spool/open_windows.go @@ -0,0 +1,32 @@ +//go:build windows + +package spool + +import ( + "errors" + "os" + + "github.com/perplexityai/numbat/internal/winfile" + "golang.org/x/sys/windows" +) + +func validateParentMode(os.FileInfo) error { return nil } + +func openExistingDatabaseFile(path string, flag int) (*os.File, error) { + if flag == os.O_RDONLY { + return winfile.OpenRegular(path) + } + return winfile.OpenExistingReadWrite(path) +} + +func validateDatabaseMode(os.FileInfo) error { return nil } + +func installDatabaseFile(candidate, path string) error { + if err := winfile.RenameNoReplace(candidate, path); err != nil { + if errors.Is(err, windows.ERROR_FILE_EXISTS) || errors.Is(err, windows.ERROR_ALREADY_EXISTS) { + return nil + } + return err + } + return nil +} diff --git a/internal/spool/spool.go b/internal/spool/spool.go new file mode 100644 index 0000000..9796e38 --- /dev/null +++ b/internal/spool/spool.go @@ -0,0 +1,348 @@ +// Package spool stores complete records until a shipper acknowledges them. +// Each operation opens, transacts against, and closes the database. A hook +// process therefore holds no database lock while it does unrelated work. +package spool + +import ( + "bytes" + "crypto/rand" + "encoding/binary" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + bolt "go.etcd.io/bbolt" + bolterrors "go.etcd.io/bbolt/errors" +) + +const ( + generationBytes = 16 + recordIDBytes = 16 + maxBatchRecords = 500 + lockTimeout = time.Second +) + +var ( + // ErrBusy reports that another process held the bbolt lock for the full + // operation timeout. A long-running shipper can retry this error. + ErrBusy = bolterrors.ErrTimeout + errNotSpool = errors.New("not a numbat spool") + + metadataBucket = []byte("numbat.spool.meta") + recordsBucket = []byte("numbat.spool.records") + markerKey = []byte("format") + markerValue = []byte("numbat-spool-v1") + generationKey = []byte("generation") + ackedKey = []byte("acked-through") +) + +// Store is a path-bound durable record queue. It owns no open resources +// between method calls and is safe to copy or use from concurrent processes. +// Its parent directory must not be writable by untrusted users. +type Store struct{ path string } + +// Batch is a FIFO prefix returned by Peek. Its acknowledgment identity is +// opaque; callers can only pass the complete batch to Ack. +type Batch struct { + Records [][]byte + generation [generationBytes]byte + boundary [recordIDBytes]byte + lastID uint64 + fileInfo os.FileInfo +} + +// New binds a store value to path without opening or creating the database. +func New(path string) Store { return Store{path: path} } + +// Put atomically appends one opaque record. A nil error means bbolt committed +// the complete value; an interrupted transaction exposes none of it. +func (store Store) Put(record []byte) error { + return store.withDB(func(db *bolt.DB, _ os.FileInfo) error { + return db.Update(func(tx *bolt.Tx) error { + records := tx.Bucket(recordsBucket) + id, err := records.NextSequence() + if err != nil { + return err + } + if id == 0 { + return errors.New("record sequence exhausted") + } + value := make([]byte, recordIDBytes+len(record)) + if _, err := rand.Read(value[:recordIDBytes]); err != nil { + return err + } + copy(value[recordIDBytes:], record) + return records.Put(key(id), value) + }) + }) +} + +// Peek returns at most 500 oldest whole records whose combined size fits +// maxBytes. It returns the first record when that record alone exceeds the +// budget. Records remain visible until Ack commits. +func (store Store) Peek(maxBytes int) (batch Batch, err error) { + if maxBytes <= 0 { + return batch, errors.New("spool: maxBytes must be positive") + } + err = store.withDB(func(db *bolt.DB, info os.FileInfo) error { + return db.View(func(tx *bolt.Tx) error { + metadata := tx.Bucket(metadataBucket) + records := tx.Bucket(recordsBucket) + copy(batch.generation[:], metadata.Get(generationKey)) + batch.fileInfo = info + size := 0 + cursor := records.Cursor() + for k, value := cursor.First(); k != nil && len(batch.Records) < maxBatchRecords; k, value = cursor.Next() { + if len(k) != 8 || len(value) < recordIDBytes { + return errors.New("invalid record entry") + } + record := value[recordIDBytes:] + if len(batch.Records) > 0 && size+len(record) > maxBytes { + break + } + batch.Records = append(batch.Records, bytes.Clone(record)) + copy(batch.boundary[:], value[:recordIDBytes]) + batch.lastID = binary.BigEndian.Uint64(k) + size += len(record) + } + return nil + }) + }) + return batch, err +} + +// Ack removes only the exact FIFO prefix represented by batch. File identity, +// store generation, and a random record identity reject replacement and +// rollback cases before any record is removed. +func (store Store) Ack(batch Batch) error { + if batch.lastID == 0 || batch.fileInfo == nil { + return errors.New("spool: cannot acknowledge an empty batch") + } + return store.withDB(func(db *bolt.DB, info os.FileInfo) error { + if !os.SameFile(batch.fileInfo, info) { + return errors.New("acknowledgment belongs to a replaced store") + } + return db.Update(func(tx *bolt.Tx) error { + metadata := tx.Bucket(metadataBucket) + records := tx.Bucket(recordsBucket) + if !bytes.Equal(metadata.Get(generationKey), batch.generation[:]) { + return errors.New("acknowledgment belongs to a replaced store") + } + acked := binary.BigEndian.Uint64(metadata.Get(ackedKey)) + if acked >= batch.lastID { + return nil + } + boundary := records.Get(key(batch.lastID)) + if len(boundary) < recordIDBytes || !bytes.Equal(boundary[:recordIDBytes], batch.boundary[:]) { + return errors.New("acknowledgment boundary no longer matches the store") + } + + last := key(batch.lastID) + keys := make([][]byte, 0, maxBatchRecords) + cursor := records.Cursor() + for k, _ := cursor.First(); k != nil && bytes.Compare(k, last) <= 0; k, _ = cursor.Next() { + if len(k) != 8 { + return errors.New("invalid record key") + } + keys = append(keys, bytes.Clone(k)) + } + for _, k := range keys { + if err := records.Delete(k); err != nil { + return err + } + } + return metadata.Put(ackedKey, key(batch.lastID)) + }) + }) +} + +func (store Store) withDB(fn func(*bolt.DB, os.FileInfo) error) (err error) { + if store.path == "" { + return errors.New("spool: empty database path") + } + parent := filepath.Dir(store.path) + if err := os.MkdirAll(parent, 0o700); err != nil { + return fmt.Errorf("spool: create parent: %w", err) + } + parentInfo, err := os.Lstat(parent) + if err != nil { + return fmt.Errorf("spool: inspect parent: %w", err) + } + if !parentInfo.IsDir() || parentInfo.Mode()&os.ModeSymlink != 0 { + return errors.New("spool: parent must be a real directory") + } + if err := validateParentMode(parentInfo); err != nil { + return fmt.Errorf("spool: parent %q: %w", parent, err) + } + + if err := ensureDatabaseFile(store.path); err != nil { + return fmt.Errorf("spool: initialize %q: %w", store.path, err) + } + file, err := openExistingDatabaseFile(store.path, os.O_RDWR) + if err != nil { + return fmt.Errorf("spool: open %q: %w", store.path, err) + } + info, err := file.Stat() + if err != nil { + _ = file.Close() + return fmt.Errorf("spool: inspect database: %w", err) + } + if !info.Mode().IsRegular() { + _ = file.Close() + return errors.New("spool: database is not a regular file") + } + if err := validateDatabaseMode(info); err != nil { + _ = file.Close() + return fmt.Errorf("spool: database %q: %w", store.path, err) + } + if err := validateDatabaseReadOnly(store.path, info); err != nil { + _ = file.Close() + return fmt.Errorf("spool: validate %q: %w", store.path, err) + } + db, err := bolt.Open(store.path, 0o600, &bolt.Options{ + Timeout: lockTimeout, + OpenFile: func(string, int, os.FileMode) (*os.File, error) { + return file, nil + }, + }) + if err != nil { + _ = file.Close() + return fmt.Errorf("spool: open %q: %w", store.path, err) + } + defer func() { err = errors.Join(err, db.Close()) }() + if err := db.View(validateStore); err != nil { + return fmt.Errorf("spool: %w", err) + } + if err := fn(db, info); err != nil { + return fmt.Errorf("spool: %w", err) + } + return nil +} + +func ensureDatabaseFile(path string) error { + if _, err := os.Lstat(path); err == nil { + return nil + } else if !errors.Is(err, os.ErrNotExist) { + return err + } + candidate, err := createDatabaseCandidate(path) + if err != nil { + return err + } + defer os.Remove(candidate) + return installDatabaseFile(candidate, path) +} + +// createDatabaseCandidate builds and closes a fully marked database beside the +// final path. An interruption can leave this private temporary file, but it +// cannot leave an unmarked database at the final path. +func createDatabaseCandidate(path string) (candidate string, err error) { + file, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+".tmp-*") + if err != nil { + return "", err + } + candidate = file.Name() + complete := false + defer func() { + _ = file.Close() + if !complete { + _ = os.Remove(candidate) + } + }() + if err := file.Chmod(0o600); err != nil { + return "", err + } + db, err := bolt.Open(candidate, 0o600, &bolt.Options{ + OpenFile: func(string, int, os.FileMode) (*os.File, error) { + return file, nil + }, + }) + if err != nil { + return "", err + } + err = db.Update(initializeStore) + if err == nil { + err = db.View(validateStore) + } + if err = errors.Join(err, db.Close()); err != nil { + return "", err + } + complete = true + return candidate, nil +} + +func validateDatabaseReadOnly(path string, want os.FileInfo) (err error) { + db, err := bolt.Open(path, 0, &bolt.Options{ + ReadOnly: true, + Timeout: lockTimeout, + OpenFile: func(path string, _ int, _ os.FileMode) (*os.File, error) { + file, err := openExistingDatabaseFile(path, os.O_RDONLY) + if err != nil { + return nil, err + } + info, err := file.Stat() + if err != nil || !os.SameFile(want, info) { + _ = file.Close() + if err != nil { + return nil, err + } + return nil, errors.New("database changed during validation") + } + return file, nil + }, + }) + if err != nil { + return err + } + defer func() { err = errors.Join(err, db.Close()) }() + return db.View(validateStore) +} + +func initializeStore(tx *bolt.Tx) error { + metadata, err := tx.CreateBucket(metadataBucket) + if err != nil { + return err + } + if _, err := tx.CreateBucket(recordsBucket); err != nil { + return err + } + generation := make([]byte, generationBytes) + if _, err := rand.Read(generation); err != nil { + return err + } + if err := metadata.Put(markerKey, markerValue); err != nil { + return err + } + if err := metadata.Put(generationKey, generation); err != nil { + return err + } + return metadata.Put(ackedKey, key(0)) +} + +func validateStore(tx *bolt.Tx) error { + metadata := tx.Bucket(metadataBucket) + records := tx.Bucket(recordsBucket) + if metadata == nil || records == nil || !bytes.Equal(metadata.Get(markerKey), markerValue) { + return errNotSpool + } + if len(metadata.Get(generationKey)) != generationBytes || len(metadata.Get(ackedKey)) != 8 { + return fmt.Errorf("%w: invalid metadata", errNotSpool) + } + if binary.BigEndian.Uint64(metadata.Get(ackedKey)) > records.Sequence() { + return fmt.Errorf("%w: invalid acknowledged sequence", errNotSpool) + } + return tx.ForEach(func(name []byte, _ *bolt.Bucket) error { + if bytes.Equal(name, metadataBucket) || bytes.Equal(name, recordsBucket) { + return nil + } + return fmt.Errorf("%w: unexpected bucket %q", errNotSpool, name) + }) +} + +func key(id uint64) []byte { + k := make([]byte, 8) + binary.BigEndian.PutUint64(k, id) + return k +} diff --git a/internal/spool/spool_diskfull_unix_test.go b/internal/spool/spool_diskfull_unix_test.go new file mode 100644 index 0000000..0e4ca2c --- /dev/null +++ b/internal/spool/spool_diskfull_unix_test.go @@ -0,0 +1,129 @@ +//go:build unix + +package spool_test + +import ( + "bytes" + "fmt" + "os" + "os/exec" + "os/signal" + "path/filepath" + "strconv" + "syscall" + "testing" + + "github.com/perplexityai/numbat/internal/spool" +) + +const ( + spoolDiskFullHelper = "NUMBAT_SPOOL_DISKFULL_HELPER" + spoolDiskFullPath = "NUMBAT_SPOOL_DISKFULL_PATH" + spoolDiskFullLimit = "NUMBAT_SPOOL_DISKFULL_LIMIT" + recordBefore = "{\"record_type\":\"event\",\"event_id\":\"before\"}\n" + recordAfter = "{\"record_type\":\"event\",\"event_id\":\"after\"}\n" +) + +// TestPutOnFullDiskCommitsNothing reproduces the failure that corrupted the +// legacy append file: a write that cannot grow its backing file. A short or +// failed append left a partial NDJSON line that the next record concatenated +// with. bbolt commits are atomic, so a Put that cannot grow the database must +// leave the store byte-identical to its pre-Put state and stay usable once space +// is available. +// +// RLIMIT_FSIZE, set in a child process so the cap cannot disturb the test +// harness, makes any file growth fail deterministically without a real full +// filesystem. A multi-megabyte record forces bbolt to grow the database past the +// cap during commit. +func TestPutOnFullDiskCommitsNothing(t *testing.T) { + if os.Getenv(spoolDiskFullHelper) == "1" { + runDiskFullPutHelper() + return + } + + path := filepath.Join(t.TempDir(), "records.spool") + store := spool.New(path) + if err := store.Put([]byte(recordBefore)); err != nil { + t.Fatalf("seed record: %v", err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat spool: %v", err) + } + + // Cap the child at the current database size so any growth during commit + // fails, then attempt a record far larger than any bbolt pre-allocation. + cmd := exec.Command(os.Args[0], "-test.run=^TestPutOnFullDiskCommitsNothing$") + cmd.Env = append(os.Environ(), + spoolDiskFullHelper+"=1", + spoolDiskFullPath+"="+path, + spoolDiskFullLimit+"="+strconv.FormatInt(info.Size(), 10), + ) + out, err := cmd.CombinedOutput() + if err == nil { + t.Fatalf("child Put unexpectedly succeeded under a file-size cap: %s", out) + } + if !bytes.Contains(out, []byte("put-failed")) { + t.Fatalf("child did not report a failed Put: %s", out) + } + + // The failed Put must have committed none of its bytes: the store still holds + // exactly the seed record, with no partial or trailing fragment. + assertSpoolRecords(t, store, recordBefore) + + // The store remains usable once space is available; the next record appends + // cleanly after the seed record rather than gluing onto a partial write. + if err := store.Put([]byte(recordAfter)); err != nil { + t.Fatalf("Put after recovered space: %v", err) + } + assertSpoolRecords(t, store, recordBefore, recordAfter) +} + +func runDiskFullPutHelper() { + // Exceeding RLIMIT_FSIZE raises SIGXFSZ, whose default action kills the + // process; ignore it so the offending write returns EFBIG instead. + signal.Ignore(syscall.SIGXFSZ) + limit, err := strconv.ParseInt(os.Getenv(spoolDiskFullLimit), 10, 64) + if err != nil { + reportDiskFullHelper("bad-limit:", err) + } + rlimit := syscall.Rlimit{Cur: uint64(limit), Max: uint64(limit)} + if err := syscall.Setrlimit(syscall.RLIMIT_FSIZE, &rlimit); err != nil { + reportDiskFullHelper("setrlimit-failed:", err) + } + record := make([]byte, 0, 4<<20) + record = append(record, []byte("{\"record_type\":\"event\",\"event_id\":\"big\",\"payload\":\"")...) + record = append(record, bytes.Repeat([]byte("a"), 4<<20)...) + record = append(record, []byte("\"}\n")...) + if err := spool.New(os.Getenv(spoolDiskFullPath)).Put(record); err != nil { + reportDiskFullHelper("put-failed:", err) + } + reportDiskFullHelper("put-succeeded", nil) +} + +// reportDiskFullHelper writes one status line the parent test matches on, then +// exits: success on the "put-succeeded" marker, failure otherwise. +func reportDiskFullHelper(marker string, cause error) { + if cause != nil { + _, _ = fmt.Fprintln(os.Stdout, marker, cause) + os.Exit(1) + } + _, _ = fmt.Fprintln(os.Stdout, marker) + os.Exit(0) +} + +func assertSpoolRecords(t *testing.T, store spool.Store, want ...string) { + t.Helper() + batch, err := store.Peek(64 << 20) + if err != nil { + t.Fatalf("peek spool: %v", err) + } + if len(batch.Records) != len(want) { + t.Fatalf("queued %d record(s), want %d: %q", len(batch.Records), len(want), batch.Records) + } + for i, record := range want { + if !bytes.Equal(batch.Records[i], []byte(record)) { + t.Fatalf("queued record %d = %q, want %q", i, batch.Records[i], record) + } + } +} diff --git a/internal/spool/spool_external_test.go b/internal/spool/spool_external_test.go new file mode 100644 index 0000000..b8e8890 --- /dev/null +++ b/internal/spool/spool_external_test.go @@ -0,0 +1,81 @@ +package spool_test + +import ( + "bufio" + "bytes" + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/perplexityai/numbat/internal/spool" +) + +const ( + spoolCommitHelper = "NUMBAT_SPOOL_COMMIT_HELPER" + spoolCommitPath = "NUMBAT_SPOOL_COMMIT_PATH" + committedRecord = "{\"record_type\":\"event\",\"event_id\":\"committed\"}\n" +) + +func TestSuccessfulPutSurvivesProducerTermination(t *testing.T) { + if os.Getenv(spoolCommitHelper) == "1" { + if err := spool.New(os.Getenv(spoolCommitPath)).Put([]byte(committedRecord)); err != nil { + t.Fatalf("Put: %v", err) + } + if _, err := fmt.Fprintln(os.Stdout, "put complete"); err != nil { + t.Fatalf("report completed Put: %v", err) + } + _, _ = bufio.NewReader(os.Stdin).ReadByte() + return + } + + path := filepath.Join(t.TempDir(), "records.spool") + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestSuccessfulPutSurvivesProducerTermination$") + cmd.Env = append(os.Environ(), spoolCommitHelper+"=1", spoolCommitPath+"="+path) + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatalf("StdoutPipe: %v", err) + } + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatalf("StdinPipe: %v", err) + } + defer stdin.Close() + var stderr bytes.Buffer + cmd.Stderr = &stderr + if err := cmd.Start(); err != nil { + t.Fatalf("start producer: %v", err) + } + waited := false + t.Cleanup(func() { + if !waited { + _ = cmd.Process.Kill() + _ = cmd.Wait() + } + }) + + line, err := bufio.NewReader(stdout).ReadString('\n') + if err != nil || line != "put complete\n" { + t.Fatalf("producer did not complete Put: line %q, error %v, stderr %q", line, err, stderr.String()) + } + if err := cmd.Process.Kill(); err != nil { + t.Fatalf("terminate producer: %v", err) + } + if err := cmd.Wait(); err == nil { + t.Fatal("terminated producer exited successfully") + } + waited = true + + batch, err := spool.New(path).Peek(1 << 20) + if err != nil { + t.Fatalf("Peek after producer termination: %v", err) + } + if len(batch.Records) != 1 || !bytes.Equal(batch.Records[0], []byte(committedRecord)) { + t.Fatalf("records after producer termination = %q, want [%q]", batch.Records, committedRecord) + } +} diff --git a/internal/winfile/winfile_windows.go b/internal/winfile/winfile_windows.go index 42e6c8a..60c7767 100644 --- a/internal/winfile/winfile_windows.go +++ b/internal/winfile/winfile_windows.go @@ -51,6 +51,13 @@ func OpenOutput(path string, appendMode bool) (*os.File, error) { return f, nil } +// OpenExistingReadWrite opens an existing regular file for bbolt. It refuses a +// reparse point at the final component. +func OpenExistingReadWrite(path string) (*os.File, error) { + access := uint32(windows.GENERIC_READ | windows.GENERIC_WRITE | windows.SYNCHRONIZE) + return open(path, access, windows.OPEN_EXISTING) +} + func open(path string, access, creation uint32) (*os.File, error) { winPath, err := extendedPath(path) if err != nil {