diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index 5e76e0f9..eb97ea23 100644 --- a/internal/cli/dataset.go +++ b/internal/cli/dataset.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "path/filepath" "github.com/spf13/cobra" "gopkg.in/yaml.v3" @@ -74,6 +75,7 @@ func newDatasetPushCmd() *cobra.Command { category string intent string labelColumn string + targetSize string // Operations flags. dryRun bool @@ -145,6 +147,7 @@ Exit codes: Context: contextOverride, Namespace: nsOverride, Spec: push.SpecArgs{Table: table, Category: category, Intent: intent, LabelColumn: labelColumn}, + TargetSizeFlag: targetSize, DryRun: dryRun, IngestorSAName: ingestorSAName, StagePodImage: stagePodImage, @@ -175,6 +178,9 @@ Exit codes: "intent: train|test") cmd.Flags().StringVar(&labelColumn, "label-column", "", "column name in labels.csv that holds the label") + cmd.Flags().StringVar(&targetSize, "target-size", "", + "image resolution as WxH (e.g. 512x512). Default: auto-detected from the first image. "+ + "All images must share this resolution — the ingestor validates it, it does not resize.") cmd.Flags().BoolVar(&dryRun, "dry-run", false, "validate + discover + walk, but don't create any cluster resources") @@ -209,6 +215,7 @@ type runDatasetPushArgs struct { Context string Namespace string Spec push.SpecArgs + TargetSizeFlag string // raw --target-size; resolved after Discover DryRun bool IngestorSAName string StagePodImage string @@ -258,11 +265,48 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush "tracebloc/client#147 non-goals.", a.Spec.Category)} } - // 3. Synthesize the spec from flags + validate against schema. + // 3. Walk the local directory FIRST. Enforces layout + size caps, + // and gives us the image list the target-size auto-detect below + // needs. Both this and the schema check are local "fail fast" + // steps; doing the walk first lets the synthesized spec carry + // the resolved target_size. + layout, err := push.Discover(a.LocalPath) + if err != nil { + return &exitError{code: 3, err: err} + } + + // 3a. Resolve the image target resolution. The ingestor's + // image_classification default is 512x512 and it VALIDATES + // (it does not resize), so a mismatch hard-fails the run with + // an "incorrect resolution" error. Honour an explicit + // --target-size; otherwise auto-detect from the first image so + // the common "all my images are NxN" case just works without + // the customer needing to know the knob exists. + if a.TargetSizeFlag != "" { + w, h, perr := push.ParseTargetSize(a.TargetSizeFlag) + if perr != nil { + return &exitError{code: 2, err: perr} + } + a.Spec.TargetSize = []int{w, h} + } else if len(layout.Images) > 0 { + if w, h, derr := push.DetectImageSize(layout.Images[0]); derr == nil { + a.Spec.TargetSize = []int{w, h} + _, _ = fmt.Fprintf(out, + "Auto-detected image target size %dx%d from %s (override with --target-size).\n", + w, h, filepath.Base(layout.Images[0])) + } else { + _, _ = fmt.Fprintf(errOut, + "Note: couldn't auto-detect image size (%v); using the ingestor "+ + "default. Pass --target-size WxH if ingestion reports a "+ + "resolution mismatch.\n", derr) + } + } + + // 4. Synthesize the spec from flags + validate against schema. // Catches "bad category", "missing intent" etc. BEFORE we - // touch the filesystem or the cluster. The error formatter - // is the same one ingest validate uses, so a customer who - // YAML'd manually first sees identical wording. + // touch the cluster. The error formatter is the same one + // ingest validate uses, so a customer who YAML'd manually + // first sees identical wording. spec := a.Spec.Build() specBytes, err := yaml.Marshal(spec) if err != nil { @@ -295,14 +339,6 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush return &exitError{code: 2, err: errors.New("synthesized spec failed schema validation; check the flag values above")} } - // 4. Walk the local directory. Enforces layout + size caps; - // customer sees a clear pointer to expected layout if they - // pass the wrong directory. - layout, err := push.Discover(a.LocalPath) - if err != nil { - return &exitError{code: 3, err: err} - } - // 5. Cluster discovery — same kubeconfig path as `cluster info`. // Errors mirror that command's exit-code contract (3 for // kubeconfig, 4 for missing release) so behaviour is @@ -527,13 +563,12 @@ func printPushPreflight( _, _ = fmt.Fprintf(out, " category: %s\n", spec["category"]) _, _ = fmt.Fprintf(out, " intent: %s\n", spec["intent"]) _, _ = fmt.Fprintf(out, " label column: %s\n", spec["label"]) - _, _ = fmt.Fprintf(out, " destination: %s\n", push.StagedPrefix(spec["table"].(string))) + _, _ = fmt.Fprintf(out, " destination: %s\n", push.FinalDestPrefix(spec["table"].(string))) _, _ = fmt.Fprintln(out) if !dryRun { - _, _ = fmt.Fprintf(out, "Next: stage %d files (%s) → %s\n", - 1+len(layout.Images), push.HumanBytes(layout.TotalBytes), - push.StagedPrefix(spec["table"].(string))) + _, _ = fmt.Fprintf(out, "Next: stage %d files (%s) for table %q\n", + 1+len(layout.Images), push.HumanBytes(layout.TotalBytes), spec["table"]) _, _ = fmt.Fprintln(out) } } diff --git a/internal/push/detect.go b/internal/push/detect.go new file mode 100644 index 00000000..9a2b838e --- /dev/null +++ b/internal/push/detect.go @@ -0,0 +1,70 @@ +package push + +import ( + "fmt" + "image" + "os" + "strconv" + "strings" + + // Register the stdlib image decoders so image.DecodeConfig can + // read the headers of the formats the image_classification layout + // accepts. webp is NOT in the stdlib — DetectImageSize returns an + // error for it and the caller falls back to requiring + // --target-size. (.jpg/.jpeg both decode via image/jpeg.) + _ "image/gif" + _ "image/jpeg" + _ "image/png" +) + +// DetectImageSize returns the pixel width and height of the image at +// path by decoding only its header (image.DecodeConfig — it does not +// read the pixel data, so it's cheap even for large images). +// +// Supports the stdlib-registered formats (jpeg, png, gif). Returns an +// error for formats without a registered decoder (notably webp); the +// caller treats that as "couldn't auto-detect" and falls back to the +// ingestor default, advising --target-size. +func DetectImageSize(path string) (width, height int, err error) { + f, err := os.Open(path) + if err != nil { + return 0, 0, err + } + defer func() { _ = f.Close() }() + + cfg, _, err := image.DecodeConfig(f) + if err != nil { + return 0, 0, fmt.Errorf("decoding image header %q: %w", path, err) + } + return cfg.Width, cfg.Height, nil +} + +// ParseTargetSize parses a --target-size flag value into [width, +// height]. Accepts "WxH" (the documented form, e.g. "512x512") and +// "W,H" as a convenience. Both dimensions must be positive integers. +func ParseTargetSize(s string) (width, height int, err error) { + sep := "x" + if strings.Contains(s, ",") { + sep = "," + } + parts := strings.Split(s, sep) + if len(parts) != 2 { + return 0, 0, fmt.Errorf( + "target size %q must be WxH (e.g. 512x512)", s) + } + width, err = strconv.Atoi(strings.TrimSpace(parts[0])) + if err != nil { + return 0, 0, fmt.Errorf( + "target size %q: width is not an integer: %w", s, err) + } + height, err = strconv.Atoi(strings.TrimSpace(parts[1])) + if err != nil { + return 0, 0, fmt.Errorf( + "target size %q: height is not an integer: %w", s, err) + } + if width <= 0 || height <= 0 { + return 0, 0, fmt.Errorf( + "target size %q: width and height must both be positive", s) + } + return width, height, nil +} diff --git a/internal/push/detect_test.go b/internal/push/detect_test.go new file mode 100644 index 00000000..14348bf3 --- /dev/null +++ b/internal/push/detect_test.go @@ -0,0 +1,88 @@ +package push + +import ( + "image" + "image/png" + "os" + "path/filepath" + "testing" +) + +// TestParseTargetSize covers the --target-size flag parser: the +// documented WxH form, the W,H convenience form, and the rejection +// cases (missing dimension, non-integer, non-positive, wrong arity). +func TestParseTargetSize(t *testing.T) { + cases := []struct { + in string + w, h int + wantErr bool + }{ + {"512x512", 512, 512, false}, + {"640x480", 640, 480, false}, + {"512,512", 512, 512, false}, + {"1x1", 1, 1, false}, + {"512", 0, 0, true}, + {"512x", 0, 0, true}, + {"x512", 0, 0, true}, + {"0x512", 0, 0, true}, + {"-4x512", 0, 0, true}, + {"512x512x512", 0, 0, true}, + {"abcxdef", 0, 0, true}, + {"", 0, 0, true}, + } + for _, c := range cases { + w, h, err := ParseTargetSize(c.in) + if c.wantErr { + if err == nil { + t.Errorf("ParseTargetSize(%q) = (%d,%d,nil), want error", c.in, w, h) + } + continue + } + if err != nil { + t.Errorf("ParseTargetSize(%q) unexpected error: %v", c.in, err) + continue + } + if w != c.w || h != c.h { + t.Errorf("ParseTargetSize(%q) = (%d,%d), want (%d,%d)", c.in, w, h, c.w, c.h) + } + } +} + +// TestDetectImageSize_PNG: a real (generated) PNG's header is decoded +// to its true dimensions. Pins the auto-detect path used when the +// customer doesn't pass --target-size. +func TestDetectImageSize_PNG(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "img.png") + f, err := os.Create(p) + if err != nil { + t.Fatal(err) + } + if err := png.Encode(f, image.NewRGBA(image.Rect(0, 0, 320, 200))); err != nil { + t.Fatal(err) + } + _ = f.Close() + + w, h, err := DetectImageSize(p) + if err != nil { + t.Fatalf("DetectImageSize: %v", err) + } + if w != 320 || h != 200 { + t.Errorf("DetectImageSize = (%d,%d), want (320,200)", w, h) + } +} + +// TestDetectImageSize_Unsupported: a non-image (or unregistered +// format) returns an error so the caller falls back to the ingestor +// default + advises --target-size, rather than silently using a +// bogus size. +func TestDetectImageSize_Unsupported(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "note.txt") + if err := os.WriteFile(p, []byte("not an image"), 0o644); err != nil { + t.Fatal(err) + } + if _, _, err := DetectImageSize(p); err == nil { + t.Error("DetectImageSize on non-image returned nil error; want a decode error") + } +} diff --git a/internal/push/spec.go b/internal/push/spec.go index 8aabb0a3..ebb7d984 100644 --- a/internal/push/spec.go +++ b/internal/push/spec.go @@ -133,6 +133,16 @@ type SpecArgs struct { // shorthand because passthrough is the only policy // image_classification cares about. LabelColumn string + + // TargetSize, when len==2, pins the image resolution as [W, H]. + // The ingestor's image_classification default is 512x512 and it + // VALIDATES (it does not resize), so a dataset whose images don't + // match the default hard-fails. Setting this emits + // spec.file_options.target_size so the customer's actual + // resolution wins. Empty (len 0) ⇒ omit and let the ingestor + // default apply. Populated by the CLI from --target-size or by + // auto-detecting the first image. + TargetSize []int } // Build produces the ingest.v1.json-conforming spec map. The @@ -157,7 +167,7 @@ type SpecArgs struct { // calls StagedPrefix, which panics on an unsafe name. func (a SpecArgs) Build() map[string]any { prefix := StagedPrefix(a.Table) - return map[string]any{ + spec := map[string]any{ "apiVersion": "tracebloc.io/v1", "kind": "IngestConfig", "category": a.Category, @@ -170,40 +180,94 @@ func (a SpecArgs) Build() map[string]any { "images": path.Join(prefix, "images") + "/", "label": a.LabelColumn, } + // Emit the image resolution under spec.file_options.target_size — + // the same override key the helm flow + data-ingestors' + // conventions.resolve honour (it merges spec.file_options over the + // per-category default). Without this, image_classification + // defaults to 512x512 and the ingestor's Image Resolution + // Validator rejects any other size. + if len(a.TargetSize) == 2 { + spec["spec"] = map[string]any{ + "file_options": map[string]any{ + "target_size": []int{a.TargetSize[0], a.TargetSize[1]}, + }, + } + } + return spec } -// StagedPrefix returns the in-cluster destination directory the CLI -// writes files into for a given table. Used in two places that -// MUST agree: +// SharedRoot is the in-cluster mount path of the chart's shared PVC +// (cluster.SharedPVCMountPath). Both the ephemeral stage Pod and the +// ingestor Job mount client-pvc here, so any path under it is visible +// to both — which is why the CLI's staging area lives under it. +const SharedRoot = "/data/shared" + +// stagingDirName is the hidden directory under SharedRoot where the +// CLI lands a run's SOURCE files. It is deliberately SEPARATE from +// the ingestor's destination (SharedRoot/): +// +// data-ingestors computes DEST_PATH = STORAGE_PATH/TABLE_NAME = +// SharedRoot/
, and its DuplicateValidator FAILS if that path +// already exists non-empty. If the CLI staged straight into +// SharedRoot/
(as it did originally), its own staging would +// create exactly the non-empty destination the validator rejects — +// so every push failed the duplicate check. Staging under +// SharedRoot/.tracebloc-staging/
keeps the destination fresh +// while remaining on the same PVC the ingestor reads. +const stagingDirName = ".tracebloc-staging" + +// StagedPrefix returns the in-cluster directory the CLI streams a +// dataset's SOURCE files into for a given table. The synthesized +// spec's csv/images point here; the ingestor reads from here and +// writes the processed table to FinalDestPrefix(table). +// +// Two call sites MUST agree on this value: // -// 1. Phase 3 (this PR + PR-b): the path the ephemeral stage Pod -// creates and tars files into. +// 1. The ephemeral stage Pod's tar target (StreamLayout). // 2. The csv/images fields in Build() above, which jobs-manager -// reads to know where the ingestor Job will find them. +// hands to the ingestor Job so it reads what we just staged. // -// Exported because Phase 3's PR-b (stage Pod construction) needs -// it from the same place, and Phase 4 (submit) might want to print -// it as part of "what we pushed." +// It is intentionally NOT SharedRoot/
: that is the ingestor's +// DEST_PATH, whose DuplicateValidator rejects a pre-existing, +// non-empty directory. See stagingDirName. // // PRECONDITION: table must already have passed ValidateTableName. // This function panics on an unsafe name rather than returning an -// escape path — a name that escapes /data/shared is a caller bug -// (validation was skipped), and a panic surfaces it loudly in -// tests instead of silently letting PR-b's stage Pod write to, -// say, /etc. Every production call path runs ValidateTableName -// first (see cli.runDatasetPush), so the panic is unreachable in -// correct code. +// escape path — a name that escapes SharedRoot is a caller bug +// (validation was skipped), and a panic surfaces it loudly in tests +// instead of silently letting the stage Pod write to, say, /etc. +// Every production call path runs ValidateTableName first (see +// cli.runDatasetPush), so the panic is unreachable in correct code. func StagedPrefix(table string) string { // Deliberately NOT path.Join here: path.Join cleans ".." // segments, which is exactly the silent traversal we're // guarding against. Plain concatenation keeps the name as a // literal segment so the assertion below can detect a bad one. - prefix := "/data/shared/" + table if !tableNamePattern.MatchString(table) { panic(fmt.Sprintf( "push.StagedPrefix: unsafe table name %q — caller must "+ "ValidateTableName before constructing a PVC path", table)) } - return prefix + return SharedRoot + "/" + stagingDirName + "/" + table +} + +// FinalDestPrefix returns where the ingestor writes the processed +// table: SharedRoot/
, matching data-ingestors' config.DEST_PATH +// (STORAGE_PATH/TABLE_NAME). This is what the training side reads and +// what the CLI shows the customer as the destination. The CLI never +// writes here directly — doing so would trip the ingestor's +// DuplicateValidator; it stages to StagedPrefix(table) and the +// ingestor produces this path. +// +// PRECONDITION: table must already have passed ValidateTableName. +// Panics on an unsafe name, same rationale as StagedPrefix. +func FinalDestPrefix(table string) string { + if !tableNamePattern.MatchString(table) { + panic(fmt.Sprintf( + "push.FinalDestPrefix: unsafe table name %q — caller must "+ + "ValidateTableName before constructing a PVC path", + table)) + } + return SharedRoot + "/" + table } diff --git a/internal/push/spec_test.go b/internal/push/spec_test.go index 05c79614..a6301cbf 100644 --- a/internal/push/spec_test.go +++ b/internal/push/spec_test.go @@ -102,8 +102,98 @@ func TestStagedPrefix_PerTableIsolation(t *testing.T) { if a, b := StagedPrefix("cats"), StagedPrefix("dogs"); a == b { t.Errorf("StagedPrefix(%q) == StagedPrefix(%q) = %q, want distinct", "cats", "dogs", a) } - if got := StagedPrefix("table_a"); got != "/data/shared/table_a" { - t.Errorf("StagedPrefix(%q) = %q, want /data/shared/table_a", "table_a", got) + if got := StagedPrefix("table_a"); got != "/data/shared/.tracebloc-staging/table_a" { + t.Errorf("StagedPrefix(%q) = %q, want /data/shared/.tracebloc-staging/table_a", "table_a", got) + } +} + +// TestStagedPrefix_DoesNotCollideWithDest is the regression pin for +// the live-discovered blocker (#26): the CLI must stage SOURCE files +// somewhere the ingestor's DEST_PATH (= FinalDestPrefix = /data/ +// shared/
) does NOT contain. If staging ever lands at or under +// the destination again, the ingestor's DuplicateValidator will +// reject the CLI's own staging as a pre-existing non-empty +// destination, and every push fails the duplicate check. +func TestStagedPrefix_DoesNotCollideWithDest(t *testing.T) { + const table = "cats_dogs_train" + staged := StagedPrefix(table) + dest := FinalDestPrefix(table) + + if staged == dest { + t.Fatalf("StagedPrefix == FinalDestPrefix == %q; the CLI would stage "+ + "into the ingestor's DEST_PATH and trip DuplicateValidator", staged) + } + // The destination must not contain the staging dir either — + // otherwise DEST is non-empty (it holds the staging subtree) and + // the validator still fails. + if strings.HasPrefix(staged, dest+"/") { + t.Errorf("StagedPrefix %q is under FinalDestPrefix %q; DEST would be "+ + "non-empty at ingest time", staged, dest) + } + if got := FinalDestPrefix(table); got != "/data/shared/"+table { + t.Errorf("FinalDestPrefix(%q) = %q, want /data/shared/%s", table, got, table) + } +} + +// TestBuild_WithTargetSize_PassesSchema pins the #27 plumbing: when +// the CLI resolves an image resolution (via --target-size or +// auto-detect), Build emits spec.file_options.target_size and the +// result still validates against the embedded v1 schema. A drift here +// would make every push that sets a target size fail validation. +func TestBuild_WithTargetSize_PassesSchema(t *testing.T) { + spec := SpecArgs{ + Table: "cats_dogs_train", + Category: "image_classification", + Intent: "train", + LabelColumn: "label", + TargetSize: []int{256, 256}, + }.Build() + + // The nested override must be present and well-shaped. + specBlock, ok := spec["spec"].(map[string]any) + if !ok { + t.Fatalf("Build() with TargetSize didn't emit a spec block: %#v", spec["spec"]) + } + fo, ok := specBlock["file_options"].(map[string]any) + if !ok { + t.Fatalf("spec.file_options missing/wrong type: %#v", specBlock["file_options"]) + } + ts, ok := fo["target_size"].([]int) + if !ok || len(ts) != 2 || ts[0] != 256 || ts[1] != 256 { + t.Fatalf("spec.file_options.target_size = %#v, want [256 256]", fo["target_size"]) + } + + specBytes, err := yaml.Marshal(spec) + if err != nil { + t.Fatalf("yaml.Marshal: %v", err) + } + v, err := schema.NewV1Validator() + if err != nil { + t.Fatalf("NewV1Validator: %v", err) + } + _, errs, parseErr := v.ValidateYAML(specBytes) + if parseErr != nil { + t.Fatalf("ValidateYAML parse error on our own output: %v\n%s", parseErr, specBytes) + } + if len(errs) != 0 { + t.Fatalf("spec with target_size failed schema validation: %s\nspec:\n%s", + schema.FormatErrors(errs), specBytes) + } +} + +// TestBuild_NoTargetSize_OmitsSpecBlock: when no resolution is set, +// Build must NOT emit a spec block (the ingestor's per-category +// default applies). Asserting the omission keeps the minimal-spec +// contract that the original schema test relies on. +func TestBuild_NoTargetSize_OmitsSpecBlock(t *testing.T) { + spec := SpecArgs{ + Table: "t", + Category: "image_classification", + Intent: "train", + LabelColumn: "label", + }.Build() + if _, present := spec["spec"]; present { + t.Errorf("Build() with no TargetSize emitted a spec block; want omitted") } } diff --git a/internal/push/stage.go b/internal/push/stage.go index f8e94725..ec396073 100644 --- a/internal/push/stage.go +++ b/internal/push/stage.go @@ -137,8 +137,8 @@ func Stage(ctx context.Context, opts StageOptions) error { // 5. Stream the tar. This is where actual bytes flow. The // progress bar (if TTY) renders during this call. - _, _ = fmt.Fprintf(opts.Out, "Streaming %d files (%s) to %s...\n", - 1+len(opts.Layout.Images), HumanBytes(opts.Layout.TotalBytes), StagedPrefix(opts.Table)) + _, _ = fmt.Fprintf(opts.Out, "Streaming %d files (%s) for table %q...\n", + 1+len(opts.Layout.Images), HumanBytes(opts.Layout.TotalBytes), opts.Table) if err := StreamLayout(ctx, opts.Executor, opts.Namespace, podName, "stage", @@ -147,7 +147,7 @@ func Stage(ctx context.Context, opts StageOptions) error { } // 6. Print "done" message. The deferred cleanup runs after this. - _, _ = fmt.Fprintf(opts.Out, "Staged %d files to %s\n", - 1+len(opts.Layout.Images), StagedPrefix(opts.Table)) + _, _ = fmt.Fprintf(opts.Out, "Staged %d files for table %q\n", + 1+len(opts.Layout.Images), opts.Table) return nil } diff --git a/internal/push/stream_test.go b/internal/push/stream_test.go index 62bde455..ea0e0553 100644 --- a/internal/push/stream_test.go +++ b/internal/push/stream_test.go @@ -187,12 +187,19 @@ func TestStreamLayout_RemoteCommand(t *testing.T) { // - tar BEFORE any rm of $DEST (preserves on tar failure) // - mv AFTER tar succeeds (atomic-ish swap) // + // `dest` is StagedPrefix(table) — the CLI's SOURCE staging dir, + // which (since #26) lives under SharedRoot/.tracebloc-staging/ so + // it never collides with the ingestor's DEST_PATH. Derive the + // expected paths from it so this test tracks StagedPrefix rather + // than hardcoding the prefix. + dest := StagedPrefix("my_table") + // Extract the staging path with a regex so the random hex // suffix doesn't pin us to a specific invocation's bytes. - stagingRE := regexp.MustCompile(`/data/shared/my_table\.staging-[0-9a-f]{8}`) + stagingRE := regexp.MustCompile(regexp.QuoteMeta(dest) + `\.staging-[0-9a-f]{8}`) stagingPaths := stagingRE.FindAllString(script, -1) if len(stagingPaths) == 0 { - t.Fatalf("remote script has no /data/shared/my_table.staging-<8hex> path (race-safety regression):\n%s", script) + t.Fatalf("remote script has no %s.staging-<8hex> path (race-safety regression):\n%s", dest, script) } // All staging mentions must refer to the SAME suffix in a single // invocation. If we see two distinct suffixes that's a bug: @@ -209,7 +216,7 @@ func TestStreamLayout_RemoteCommand(t *testing.T) { for _, want := range []string{ `mkdir -p "` + staging + `"`, `tar -xf - -C "` + staging + `"`, - `mv "` + staging + `" "/data/shared/my_table"`, + `mv "` + staging + `" "` + dest + `"`, } { if !strings.Contains(script, want) { t.Errorf("remote script missing %q: %s", want, script) @@ -221,18 +228,24 @@ func TestStreamLayout_RemoteCommand(t *testing.T) { // (backup-and-swap, so a mv failure can be rolled back). // The contract is the same: tar runs while $DEST is intact. tarIdx := strings.Index(script, `tar -xf - -C "`+staging+`"`) - destBackupMvIdx := strings.Index(script, `mv "/data/shared/my_table" "`) + destBackupMvIdx := strings.Index(script, `mv "`+dest+`" "`) if tarIdx < 0 || destBackupMvIdx < 0 { t.Fatalf("remote script missing tar or destination-backup mv: %s", script) } if tarIdx >= destBackupMvIdx { t.Errorf("remote script touches $DEST BEFORE tar succeeds — partial-transfer could destroy previous data:\n%s", script) } - // Single-segment guarantee: both rm targets must end with - // /my_table or /my_table.staging-* — never just /data/shared. - if strings.Contains(script, `rm -rf "/data/shared"`) || - strings.Contains(script, `rm -rf "/data/shared/"`) { - t.Errorf("remote script rm-rfs the parent /data/shared (would nuke sibling tables):\n%s", script) + // Single-segment guarantee: rm targets must never be the shared + // root or the staging parent themselves (that would nuke sibling + // tables / every in-flight push). + for _, forbidden := range []string{ + `rm -rf "/data/shared"`, + `rm -rf "/data/shared/"`, + `rm -rf "` + SharedRoot + "/" + stagingDirName + `"`, + } { + if strings.Contains(script, forbidden) { + t.Errorf("remote script contains dangerous %q (would nuke sibling tables/pushes):\n%s", forbidden, script) + } } // Bugbot r9 + r10: orphan cleanup for previously-failed pushes @@ -253,7 +266,7 @@ func TestStreamLayout_RemoteCommand(t *testing.T) { // must be backed up to .old- BEFORE the new dataset // arrives, and restored if the main mv fails. Pin the key // shape pieces — backup mv, primary mv, rollback mv, cleanup. - backupRE := regexp.MustCompile(`/data/shared/my_table\.old-[0-9a-f]{8}`) + backupRE := regexp.MustCompile(regexp.QuoteMeta(dest) + `\.old-[0-9a-f]{8}`) backupPaths := backupRE.FindAllString(script, -1) if len(backupPaths) == 0 { t.Fatalf("remote script has no .old- backup path (r10 rollback regression):\n%s", script) @@ -272,15 +285,15 @@ func TestStreamLayout_RemoteCommand(t *testing.T) { // suffixes would defeat the "find -name ...staging-* ...old-*" // orphan-cleanup symmetry, AND would risk collision with a // concurrent push's .old-. - if strings.TrimPrefix(backup, "/data/shared/my_table.old-") != - strings.TrimPrefix(staging, "/data/shared/my_table.staging-") { + if strings.TrimPrefix(backup, dest+".old-") != + strings.TrimPrefix(staging, dest+".staging-") { t.Errorf("backup and staging suffixes diverge: %q vs %q", backup, staging) } // Backup mv (DEST → .old) must appear BEFORE primary mv // (.staging → DEST), or rollback wouldn't have anything to // restore. - backupMvIdx := strings.Index(script, `mv "/data/shared/my_table" "`+backup+`"`) - primaryMvIdx := strings.Index(script, `mv "`+staging+`" "/data/shared/my_table"`) + backupMvIdx := strings.Index(script, `mv "`+dest+`" "`+backup+`"`) + primaryMvIdx := strings.Index(script, `mv "`+staging+`" "`+dest+`"`) if backupMvIdx < 0 || primaryMvIdx < 0 { t.Fatalf("remote script missing backup or primary mv:\n%s", script) } @@ -299,7 +312,7 @@ func TestStreamLayout_StagingSuffixIsUniquePerInvocation(t *testing.T) { if err != nil { t.Fatalf("Discover: %v", err) } - stagingRE := regexp.MustCompile(`/data/shared/t\.staging-[0-9a-f]{8}`) + stagingRE := regexp.MustCompile(regexp.QuoteMeta(StagedPrefix("t")) + `\.staging-[0-9a-f]{8}`) collect := func() string { fe := &fakeExecutor{} diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 3280e813..3ef8d477 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -220,18 +220,10 @@ func WatchJob( // a structured representation of the banner without // requiring a second log fetch post-completion. summary, logErr := streamPodLogsAndParse(watchCtx, cs, namespace, podName, out) - // Filter out the two ctx-flavored errors — both are "observation - // gave up early," not "stream failed." They get classified below - // into Detached (customer SIGINT, JobWatchTimeout expiry). Any - // other error is a real streaming failure (network mid-stream, - // API server tantrum) and bubbles up as a watch error. - if logErr != nil && - !errors.Is(logErr, context.Canceled) && - !errors.Is(logErr, context.DeadlineExceeded) { - return nil, fmt.Errorf("streaming logs from Pod %s/%s: %w", namespace, podName, logErr) - } - // 3. Detach branches: + // 3. Detach branches — checked FIRST, since the customer's SIGINT + // or the watch-cap expiry is the operative intent and takes + // precedence over any stream error: // - customerCtx canceled = SIGINT // - watchCtx expired (DeadlineExceeded) = JobWatchTimeout cap // hit during streaming (1-hour observation window exceeded) @@ -265,20 +257,54 @@ func WatchJob( // inheriting watchCtx's depleted budget caused successful // slow ingestions to misreport as Unknown. // - // The fresh ctx still propagates SIGINT (parent is - // customerCtx, which carries signal.NotifyContext's - // cancel). If the customer Ctrl-C's during this 30s - // window, we fall into the detach branch below — same - // contract as during the log stream. + // The Job — not the log stream — is the source of truth for + // success/failure, so we ALWAYS consult it here, INCLUDING when + // the log stream broke for a non-ctx reason (#28: the watched + // Pod was replaced / restarted / deleted mid-follow, e.g. a + // backoffLimit retry). A broken stream is only fatal if we also + // can't determine the Job's outcome. + // + // The fresh ctx still propagates SIGINT (parent is customerCtx, + // which carries signal.NotifyContext's cancel); a Ctrl-C in this + // window falls into the detach branches below. finalCtx, finalCancel := context.WithTimeout(customerCtx, 30*time.Second) defer finalCancel() - outcome, err := finalJobStatus(finalCtx, cs, namespace, jobName) - if err != nil { + outcome, statusErr := finalJobStatus(finalCtx, cs, namespace, jobName) + + // A non-ctx log-stream error is incidental if the Job still + // reached a terminal state. Previously ANY such error (e.g. + // "container is terminated" once the Pod was replaced by a retry) + // returned exit 9 even when the Job ultimately succeeded. #28. + streamFailed := logErr != nil && + !errors.Is(logErr, context.Canceled) && + !errors.Is(logErr, context.DeadlineExceeded) + if streamFailed { + // SIGINT during the final-status poll → graceful detach. + if errors.Is(customerCtx.Err(), context.Canceled) { + return &WatchResult{ + Outcome: JobOutcomeDetached, + PodName: podName, + Summary: summary, + DetachReason: DetachReasonSignal, + }, nil + } + // Job reached a terminal state → the stream error was + // incidental (Pod replaced/restarted). Report the real + // outcome the customer cares about. + if statusErr == nil && (outcome == JobOutcomeSucceeded || outcome == JobOutcomeFailed) { + return &WatchResult{Outcome: outcome, PodName: podName, Summary: summary}, nil + } + // Couldn't confirm a terminal Job state → the stream failure + // is the actionable signal; surface it. + return nil, fmt.Errorf("streaming logs from Pod %s/%s: %w", namespace, podName, logErr) + } + + // 5. Clean-stream path: classify on the Job status alone. + if statusErr != nil { // Treat SIGINT during finalJobStatus as graceful detach - // (same as during the log stream — jobs-manager already - // accepted the run, the customer is just stopping the - // observation). Bugbot PR #10 r2 flagged the "exit 9 on - // post-stream SIGINT" inconsistency. + // (jobs-manager already accepted the run; the customer is + // just stopping the observation). Bugbot PR #10 r2 flagged + // the "exit 9 on post-stream SIGINT" inconsistency. if errors.Is(customerCtx.Err(), context.Canceled) { return &WatchResult{ Outcome: JobOutcomeDetached, @@ -287,7 +313,7 @@ func WatchJob( DetachReason: DetachReasonSignal, }, nil } - return nil, fmt.Errorf("reading final Job status for %s/%s: %w", namespace, jobName, err) + return nil, fmt.Errorf("reading final Job status for %s/%s: %w", namespace, jobName, statusErr) } return &WatchResult{ Outcome: outcome, diff --git a/internal/submit/watch_test.go b/internal/submit/watch_test.go index 2be8b052..4da6fb9a 100644 --- a/internal/submit/watch_test.go +++ b/internal/submit/watch_test.go @@ -263,6 +263,62 @@ func TestWatchJob_PodWaitTimeoutMapsToDetach(t *testing.T) { } } +// TestWatchJob_TerminalJobStatusWins is the #28 regression pin: the +// Job — not the log stream — is the source of truth for the outcome. +// With a Running Pod and a Job already reporting Complete, WatchJob +// must return Succeeded. This holds whether the (fake) log stream +// yields data or breaks: if the stream errored, the new +// "streamFailed but Job terminal → report outcome" branch still +// resolves to the Job's verdict instead of bubbling exit-9. Before +// the fix, a broken stream (e.g. a Pod replaced by a retry, or +// deleted mid-follow) returned an error even on a successful Job. +func TestWatchJob_TerminalJobStatusWins(t *testing.T) { + cs := fake.NewClientset( + jobPod("ingestor-xyz", "ingestor", corev1.PodRunning), + jobWithCondition("ingestor", batchv1.JobComplete), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var out bytes.Buffer + wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor", &out) + if err != nil { + t.Fatalf("WatchJob returned error; want nil + Succeeded: %v", err) + } + if wr == nil { + t.Fatal("WatchJob returned nil result") + } + if wr.Outcome != JobOutcomeSucceeded { + t.Fatalf("Outcome = %v, want Succeeded (Job condition is the source of truth)", wr.Outcome) + } +} + +// TestWatchJob_TerminalFailedJobReported: the mirror of the above for +// a Failed Job — the watch reports Failed (→ exit 9), not a generic +// watch error, even if the stream broke. +func TestWatchJob_TerminalFailedJobReported(t *testing.T) { + cs := fake.NewClientset( + jobPod("ingestor-xyz", "ingestor", corev1.PodRunning), + jobWithCondition("ingestor", batchv1.JobFailed), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var out bytes.Buffer + wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor", &out) + if err != nil { + t.Fatalf("WatchJob returned error; want nil + Failed: %v", err) + } + if wr == nil { + t.Fatal("WatchJob returned nil result") + } + if wr.Outcome != JobOutcomeFailed { + t.Fatalf("Outcome = %v, want Failed", wr.Outcome) + } +} + // TestJobOutcome_String: stringer pin so diagnostic output stays // stable. func TestJobOutcome_String(t *testing.T) {