From 69c71292c2121e0ef5a81d7f8b192bf1a2a2039d Mon Sep 17 00:00:00 2001 From: Lukas Wuttke Date: Thu, 9 Jul 2026 20:32:49 +0200 Subject: [PATCH 1/4] fix(auth): reject unknown --env/$CLIENT_ENV at login instead of silently using prod MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BaseURL falls unknown/typo env values back to prod (a lenient library default), so `login --env staging` or `CLIENT_ENV=prd` silently targeted production AND persisted it as the active session env for every later command — the class behind earlier dev-vs-prod confusion. login PICKS and persists the session env, so a typo must fail there. Add api.IsKnownEnv (dev/stg/prod, case-insensitive) and validate the resolved env at runLogin entry, before any network call. ResolveEnv still maps empty->prod, so the no-flag default is unaffected. BaseURL's unknown->prod fallback is deliberately unchanged (TestBaseURL asserts it). Co-Authored-By: Claude Opus 4.8 --- internal/api/client.go | 15 +++++++++++++++ internal/api/client_test.go | 17 +++++++++++++++++ internal/cli/auth.go | 9 +++++++++ 3 files changed, 41 insertions(+) diff --git a/internal/api/client.go b/internal/api/client.go index cc7e958b..15a95bd5 100644 --- a/internal/api/client.go +++ b/internal/api/client.go @@ -106,6 +106,21 @@ func ResolveEnv(explicit string) string { return EnvProd } +// IsKnownEnv reports whether env is one of the recognized backends (dev/stg/prod, +// case-insensitively). Callers that let a human PICK the env (e.g. `login`) use it +// to reject a typo up front — BaseURL deliberately falls unknown values back to +// prod (a lenient library default), so without this a `--env staging`/`prd` typo +// would silently target production. Empty is NOT known here: resolve first +// (ResolveEnv turns empty into the prod default), then validate the result. +func IsKnownEnv(env string) bool { + switch strings.ToLower(env) { + case EnvDev, EnvStg, EnvProd: + return true + default: + return false + } +} + // Client talks to the backend REST API. Token (the user token from login) is // optional: the device-flow endpoints are unauthenticated; provisioning calls // set it. diff --git a/internal/api/client_test.go b/internal/api/client_test.go index 11ba575a..2a1f107a 100644 --- a/internal/api/client_test.go +++ b/internal/api/client_test.go @@ -41,6 +41,23 @@ func TestResolveEnv(t *testing.T) { } } +func TestIsKnownEnv(t *testing.T) { + known := []string{"dev", "stg", "prod", "DEV", "Prod"} // case-insensitive + for _, env := range known { + if !IsKnownEnv(env) { + t.Errorf("IsKnownEnv(%q) = false, want true", env) + } + } + // Typos and the unknown values BaseURL would silently route to prod + // must be rejected so `login` fails instead of persisting a prod session. + unknown := []string{"staging", "prd", "production", "development", "", " "} + for _, env := range unknown { + if IsKnownEnv(env) { + t.Errorf("IsKnownEnv(%q) = true, want false", env) + } + } +} + func TestRequestDeviceCode(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/device/code" || r.Method != http.MethodPost { diff --git a/internal/cli/auth.go b/internal/cli/auth.go index f408aca7..c8c65cd3 100644 --- a/internal/cli/auth.go +++ b/internal/cli/auth.go @@ -56,6 +56,15 @@ func runLogin(ctx context.Context, p *ui.Printer, envFlag string) error { return &exitError{code: 1, err: err} } env := api.ResolveEnv(envFlag) + // login PICKS the session env and persists it (cfg.CurrentEnv below), so a + // typo must fail HERE, not silently resolve to prod. BaseURL's lenient + // unknown→prod fallback would otherwise route `--env staging` / `CLIENT_ENV=prd` + // to production and store it as the active env for every later command. + if !api.IsKnownEnv(env) { + return &exitError{code: 1, err: fmt.Errorf( + "unknown backend environment %q — valid values are dev, stg, prod (default). "+ + "Check --env / $CLIENT_ENV", env)} + } client := newAPIClient(env) p.Detailf("backend %s — requesting a device code …", client.BaseURL) From cc019352af7c2d58210d23ed527d61fc21270f45 Mon Sep 17 00:00:00 2001 From: Lukas Wuttke Date: Thu, 9 Jul 2026 20:33:03 +0200 Subject: [PATCH 2/4] fix(submit): raise log-scanner buffer to 16 MB so tqdm progress can't force a false exit 9 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tqdm (a data-ingestors dep) redraws its progress bar with \r and no \n, so a whole ingestion phase's redraws are one newline-delimited "line" that grows for the life of the run. Past 1 MB the display scanner returned bufio.ErrTooLong, cutting the log stream mid-run; a still-running Job then couldn't be confirmed terminal in the 30s finalJobStatus poll, so watch returned a false exit 9 on a healthy large ingestion — exactly the case the 1h JobWatchTimeout targets. The parser is fed via the TeeReader, not the scanner, so this cap only ever bounded the DISPLAY line and never the verdict. Raise it to 16 MB (clears a fast ~10/s hour of redraws with headroom; the 1h cap bounds accumulation). The buffer grows on demand, so ordinary log lines still cost 64 KB. Co-Authored-By: Claude Opus 4.8 --- internal/submit/watch.go | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/internal/submit/watch.go b/internal/submit/watch.go index bc5a6ab2..7fdbc442 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -464,11 +464,18 @@ func streamPodLogsAndParse( // transport layer, making the output feel laggy on a fast // ingestion. scanner := bufio.NewScanner(tee) - // Default scanner buffer is 64 KB per line — fine for log - // lines but bump to 1 MB to handle the (rare) case where a - // single ingestion-error stacktrace has a multi-KB Python - // traceback line. - scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + // The scanner splits the DISPLAY stream on '\n' (the parser is fed + // separately via the TeeReader, so this cap never affects the verdict). + // tqdm — a data-ingestors dep — redraws its progress bar with '\r' and NO + // '\n', so a whole ingestion phase's worth of redraws is a single + // newline-delimited "line" that grows for the life of the run. At 1 MB that + // overflowed on a big ingestion (ErrTooLong cut the stream mid-run, and a + // still-running Job then couldn't be confirmed terminal in the 30s + // finalJobStatus poll → a false exit 9). The watch is capped at 1h + // (JobWatchTimeout), which bounds the accumulation; 16 MB clears a fast + // (~10/s) hour of redraws with headroom. The buffer grows on demand, so + // ordinary log lines still cost 64 KB. + scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) for scanner.Scan() { line := scanner.Bytes() // Print the line + a newline (scanner strips the trailing From 236ec6848385b1b7ee7ea88ce65fab571717561d Mon Sep 17 00:00:00 2001 From: Lukas Wuttke Date: Fri, 10 Jul 2026 12:44:35 +0200 Subject: [PATCH 3/4] fix(submit): drain past ErrTooLong so a giant tqdm line can't force a false exit 9 (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses @saadqbal's review on #208: the 16 MB buffer bump only MOVED the false-exit-9 threshold, it didn't close it. The tee is pulled only by the DISPLAY scanner, so when a line trips ErrTooLong the scan loop exits, the tee stops being read, and the parser never sees the rest of the stream (the closing banner) → streamFailed && outcome==Unknown → a false exit 9 on a healthy run. A long enough single '\r'-line (> the buffer) still breaks it. Class-level fix (his suggestion): keep draining past ErrTooLong. Extracted the display/parse loop into streamDisplayAndParse; on ErrTooLong it drains the rest of the stream THROUGH the tee (io.Copy to io.Discard) so the parser still sees the banner, and it is NOT fatal — the Job status poll is the verdict's source of truth. Genuine read failures (network drop, ctx cancel) still propagate. Corrected the now-wrong "cap never affects the verdict" comment; kept 16 MB as a generous display headroom (the drain is the correctness guarantee). New tests (the #3 no-test gap Asad noted): an oversized '\r'-line + the real ingestor banner → the parser still resolves the summary (no false exit 9); and a genuine read error still propagates. Co-Authored-By: Claude Opus 4.8 --- internal/submit/watch.go | 83 +++++++++++++++++---------- internal/submit/watch_display_test.go | 67 +++++++++++++++++++++ 2 files changed, 120 insertions(+), 30 deletions(-) create mode 100644 internal/submit/watch_display_test.go diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 7fdbc442..6f8dfa66 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -463,36 +463,16 @@ func streamPodLogsAndParse( // io.Copy would also work but would buffer chunks at the // transport layer, making the output feel laggy on a fast // ingestion. - scanner := bufio.NewScanner(tee) - // The scanner splits the DISPLAY stream on '\n' (the parser is fed - // separately via the TeeReader, so this cap never affects the verdict). - // tqdm — a data-ingestors dep — redraws its progress bar with '\r' and NO - // '\n', so a whole ingestion phase's worth of redraws is a single - // newline-delimited "line" that grows for the life of the run. At 1 MB that - // overflowed on a big ingestion (ErrTooLong cut the stream mid-run, and a - // still-running Job then couldn't be confirmed terminal in the 30s - // finalJobStatus poll → a false exit 9). The watch is capped at 1h - // (JobWatchTimeout), which bounds the accumulation; 16 MB clears a fast - // (~10/s) hour of redraws with headroom. The buffer grows on demand, so - // ordinary log lines still cost 64 KB. - scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) - for scanner.Scan() { - line := scanner.Bytes() - // Print the line + a newline (scanner strips the trailing - // '\n'). errcheck-friendly: we discard the writer error - // because the exit code is the customer-facing contract. - _, _ = out.Write(line) - _, _ = out.Write([]byte("\n")) - } - if err := scanner.Err(); err != nil { - // EOF is normal end-of-stream; other errors (network drop - // mid-stream, ctx cancel) get propagated. - if !errors.Is(err, io.EOF) { - // Flush any buffered partial line before returning so - // the parser sees content even on mid-line failure. - parser.FlushLine() - return parser.Result(), err - } + // Pull the whole stream through the tee (which feeds the summary parser) + // while rendering it line-by-line for the customer. An over-long DISPLAY + // line (a tqdm '\r'-redraw burst) is drained, not fatal — see + // streamDisplayAndParse. + if err := streamDisplayAndParse(tee, out, displayLineMax); err != nil { + // A genuine mid-stream failure (network drop, ctx cancel). Flush any + // buffered partial line so the parser sees content even on a mid-line + // failure, then propagate. + parser.FlushLine() + return parser.Result(), err } // Flush at the end of stream too. A Pod that exited without // a trailing newline on its final stdout write would otherwise @@ -502,6 +482,49 @@ func streamPodLogsAndParse( return parser.Result(), nil } +// displayLineMax caps the per-line DISPLAY buffer (grows on demand from 64 KB, +// so ordinary log lines still cost 64 KB). A tqdm progress "line" — many '\r' +// redraws with no '\n' — grows for the life of the run; the generous cap lets +// the common case render fully, and streamDisplayAndParse drains anything past +// it rather than failing. The 1h JobWatchTimeout bounds accumulation. +const displayLineMax = 16 * 1024 * 1024 + +// streamDisplayAndParse renders r line-by-line to out for the customer. The +// caller wraps the log stream in an io.TeeReader that ALSO feeds the summary +// parser, and this is the ONLY place the tee is pulled — so the function's real +// job is to pull the WHOLE stream through, whatever the display does with it. +// +// A tqdm progress "line" (many '\r' redraws, no '\n') can outgrow maxLine. That +// must NOT stop the pull: if it did, the tee would stop feeding the parser and +// the closing banner would never be parsed, so watch would report a false exit +// 9 on a healthy run (raising maxLine only postpones the threshold). On +// bufio.ErrTooLong we therefore keep draining the rest of the stream — +// discarding only the oversized DISPLAY line — so the parser still sees the +// banner; the Job status poll is the verdict's real source of truth. Returns a +// non-nil error only on a genuine read failure (network drop, ctx cancel). +func streamDisplayAndParse(r io.Reader, out io.Writer, maxLine int) error { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), maxLine) + for scanner.Scan() { + // scanner strips the trailing '\n'; re-add it. errcheck-friendly: the + // write error is discarded because the exit code is the contract. + _, _ = out.Write(scanner.Bytes()) + _, _ = out.Write([]byte("\n")) + } + err := scanner.Err() + if err == nil || errors.Is(err, io.EOF) { + return nil + } + if errors.Is(err, bufio.ErrTooLong) { + // Keep draining THROUGH r (the tee) so the parser still sees the banner. + if _, derr := io.Copy(io.Discard, r); derr != nil { + return derr + } + return nil + } + return err +} + // parserWriter adapts a SummaryParser into an io.Writer for use // with io.TeeReader. The TeeReader writes everything to its // secondary sink as bytes flow through; the parser's Feed method diff --git a/internal/submit/watch_display_test.go b/internal/submit/watch_display_test.go new file mode 100644 index 00000000..6bf86c96 --- /dev/null +++ b/internal/submit/watch_display_test.go @@ -0,0 +1,67 @@ +package submit + +import ( + "bytes" + "errors" + "io" + "strings" + "testing" +) + +// errAfterReader yields its data, then returns err on the next Read — a genuine +// mid-stream read failure (network drop / ctx cancel), distinct from io.EOF. +type errAfterReader struct { + data []byte + err error + pos int +} + +func (r *errAfterReader) Read(p []byte) (int, error) { + if r.pos >= len(r.data) { + return 0, r.err + } + n := copy(p, r.data[r.pos:]) + r.pos += n + return n, nil +} + +// TestStreamDisplayAndParse_DrainsPastOversizedLineSoParserSeesBanner pins the +// #208-review fix: an over-long tqdm-style progress "line" (many '\r' redraws, +// no '\n') that outgrows the display buffer, immediately followed by the real +// closing banner. Because the tee is pulled ONLY by the display scanner, a +// naive ErrTooLong bail would stop the parser from ever seeing the banner → +// watch returns a false exit 9 on a healthy run. The drain-past-ErrTooLong must +// keep pulling so the parser still resolves the summary. +func TestStreamDisplayAndParse_DrainsPastOversizedLineSoParserSeesBanner(t *testing.T) { + oversized := strings.Repeat("\rprocessing... ", 500) // ~7 KB, no '\n' + stream := strings.NewReader(oversized + realIngestorBanner) + parser := NewSummaryParser() + tee := io.TeeReader(stream, parserWriter{parser: parser}) + + var out bytes.Buffer + // Tiny cap so the oversized line trips ErrTooLong (production uses 16 MB). + if err := streamDisplayAndParse(tee, &out, 1024); err != nil { + t.Fatalf("an over-long DISPLAY line must not be fatal; got: %v", err) + } + parser.FlushLine() + if got := parser.Result().InsertedRecords; got != 1200 { + t.Fatalf( + "parser missed the banner after the oversized line (false exit 9): "+ + "InsertedRecords=%d, want 1200", + got, + ) + } +} + +// TestStreamDisplayAndParse_GenuineReadErrorPropagates: a real mid-stream read +// failure (not ErrTooLong, not EOF) stays fatal — the drain path must not +// swallow genuine stream errors. +func TestStreamDisplayAndParse_GenuineReadErrorPropagates(t *testing.T) { + want := errors.New("connection reset by peer") + r := &errAfterReader{data: []byte("some log line\n"), err: want} + + var out bytes.Buffer + if err := streamDisplayAndParse(r, &out, 1024); !errors.Is(err, want) { + t.Fatalf("a genuine read error should propagate; got: %v", err) + } +} From 38d92db87477ad559da321108fa5cb69630181af Mon Sep 17 00:00:00 2001 From: Lukas Wuttke Date: Fri, 10 Jul 2026 12:52:14 +0200 Subject: [PATCH 4/4] =?UTF-8?q?test(push):=20pin=20Inf/NaN=20=E2=86=92=20n?= =?UTF-8?q?ot=20FLOAT=20against=20the=20di#349=20inference?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #208's fix #2 (demote Inf/NaN columns to VARCHAR) is superseded by #185/#210: the di#349 floatRE grammar pre-screens the token before ParseFloat, so "inf"/"Infinity"/"NaN" already fall through to VARCHAR. Dropped the redundant production change on rebase; kept the intent as a regression test, since no parity-fixture case covers non-finite. Co-Authored-By: Claude Opus 4.8 --- internal/push/tabular_test.go | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/internal/push/tabular_test.go b/internal/push/tabular_test.go index f9e3440b..ce10180f 100644 --- a/internal/push/tabular_test.go +++ b/internal/push/tabular_test.go @@ -167,6 +167,27 @@ func TestInferSchema(t *testing.T) { } } +// TestInferSchema_NonFiniteIsNotFloat: Go's strconv.ParseFloat accepts +// "Inf"/"Infinity"/"NaN", but the ingestor's FLOAT cast rejects a non-finite +// value — so a column carrying one must NOT infer FLOAT (that would hand the +// cluster a schema it only refuses AFTER the upload). The di#349 float grammar +// (floatRE) pre-screens the token before ParseFloat, so "inf"/"NaN" fall +// through to VARCHAR and preflight matches what the cluster will accept. No +// parity-fixture case covers this, so pin it here. +func TestInferSchema_NonFiniteIsNotFloat(t *testing.T) { + dir := t.TempDir() + csv := writeFile(t, dir, "data.csv", + "reading,note\n1.5,ok\ninf,spike\nNaN,dropout\n") + res, err := InferSchema(csv) + if err != nil { + t.Fatalf("InferSchema: %v", err) + } + // Longest of 1.5/inf/NaN is 3 runes → VARCHAR(3); the point is it is NOT FLOAT. + if got := res.Schema["reading"]; got != "VARCHAR(3)" { + t.Errorf("schema[reading] = %q, want VARCHAR(3) (Inf/NaN must not infer FLOAT)", got) + } +} + // TestInferSchema_EmptyColumnIsVarchar1: a column with no non-empty sampled // value can't be typed from data; it comes back as VARCHAR(1) (mirroring // the ingestor's all-missing rule) and is reported in the Empty list so