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) 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 diff --git a/internal/submit/watch.go b/internal/submit/watch.go index bc5a6ab2..6f8dfa66 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -463,29 +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) - // 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) - 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 @@ -495,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) + } +}