From d74277552f4bf5432f1ebf228135556cdb83ce4a Mon Sep 17 00:00:00 2001 From: Bakhtiyar Issakhmetov Date: Sat, 5 Sep 2026 13:49:27 +0500 Subject: [PATCH] fix(landscape): bound node fan-out, cover override resolution, test the run consumer Co-Authored-By: Claude Fable 5.1 --- .../messaging/rabbitmq/landscape_consumer.go | 7 +- .../rabbitmq/landscape_consumer_test.go | 90 +++++++++++++++++++ internal/service/landscape.go | 36 +++++++- internal/service/landscape_test.go | 63 ++++++++++++- 4 files changed, 188 insertions(+), 8 deletions(-) create mode 100644 internal/messaging/rabbitmq/landscape_consumer_test.go diff --git a/internal/messaging/rabbitmq/landscape_consumer.go b/internal/messaging/rabbitmq/landscape_consumer.go index 96350e6..bf3c5d9 100644 --- a/internal/messaging/rabbitmq/landscape_consumer.go +++ b/internal/messaging/rabbitmq/landscape_consumer.go @@ -67,8 +67,11 @@ func (h landscapeHandler) Handle(ctx context.Context, d amqp.Delivery) error { if err != nil { // The delivery is nacked without requeue (see Consumer.dispatch), so the // run will not be retried. Record the failure on the run itself; - // otherwise it stays "running" forever. Best-effort: log if marking - // fails, but still return the original error so the delivery is nacked. + // otherwise it stays "running" forever. This covers the errors Execute + // returns before it can mark the run itself (a missing run, a failed + // status write), since Execute already marks the run failed for every + // node failure. Best-effort: log if marking fails, but still return the + // original error so the delivery is nacked. if markErr := h.proc.MarkRunFailed(ctx, job.RunID, err.Error()); markErr != nil { h.log.Error("mark landscape run failed", "run_id", job.RunID, "error", markErr) } diff --git a/internal/messaging/rabbitmq/landscape_consumer_test.go b/internal/messaging/rabbitmq/landscape_consumer_test.go new file mode 100644 index 0000000..1487dee --- /dev/null +++ b/internal/messaging/rabbitmq/landscape_consumer_test.go @@ -0,0 +1,90 @@ +package rabbitmq + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/google/uuid" + amqp "github.com/rabbitmq/amqp091-go" +) + +// fakeLandscapeProcessor records MarkRunFailed calls and returns a preset error +// from Execute. +type fakeLandscapeProcessor struct { + runErr error + executed []uuid.UUID + failedID uuid.UUID + failedWith string + failedCnt int +} + +func (f *fakeLandscapeProcessor) Execute(_ context.Context, runID uuid.UUID) error { + f.executed = append(f.executed, runID) + return f.runErr +} + +func (f *fakeLandscapeProcessor) MarkRunFailed(_ context.Context, runID uuid.UUID, reason string) error { + f.failedCnt++ + f.failedID = runID + f.failedWith = reason + return nil +} + +func landscapeDelivery(t *testing.T, id uuid.UUID) amqp.Delivery { + t.Helper() + body, err := json.Marshal(LandscapeJob{RunID: id}) + if err != nil { + t.Fatalf("marshal job: %v", err) + } + return amqp.Delivery{Body: body} +} + +func TestLandscapeHandler_MarksRunFailedOnExecuteError(t *testing.T) { + id := uuid.New() + fp := &fakeLandscapeProcessor{runErr: errors.New("node 00: grid mismatch")} + h := NewLandscapeRunHandler(fp, discardLogger()) + + err := h.Handle(context.Background(), landscapeDelivery(t, id)) + if err == nil { + t.Fatal("expected the run error to be returned so the delivery is nacked") + } + if len(fp.executed) != 1 || fp.executed[0] != id { + t.Fatalf("Execute called with %v, want [%s]", fp.executed, id) + } + if fp.failedCnt != 1 { + t.Fatalf("expected MarkRunFailed to be called once, got %d", fp.failedCnt) + } + if fp.failedID != id { + t.Fatalf("MarkRunFailed called with wrong id: got %s want %s", fp.failedID, id) + } + if fp.failedWith != "node 00: grid mismatch" { + t.Fatalf("MarkRunFailed called with wrong reason: %q", fp.failedWith) + } +} + +func TestLandscapeHandler_DoesNotMarkFailedOnSuccess(t *testing.T) { + fp := &fakeLandscapeProcessor{runErr: nil} + h := NewLandscapeRunHandler(fp, discardLogger()) + + if err := h.Handle(context.Background(), landscapeDelivery(t, uuid.New())); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if fp.failedCnt != 0 { + t.Fatalf("expected MarkRunFailed not to be called on success, got %d", fp.failedCnt) + } +} + +func TestLandscapeHandler_UndecodableBodyIsNacked(t *testing.T) { + fp := &fakeLandscapeProcessor{} + h := NewLandscapeRunHandler(fp, discardLogger()) + + err := h.Handle(context.Background(), amqp.Delivery{Body: []byte("not json")}) + if err == nil { + t.Fatal("expected a decode error so the delivery is nacked") + } + if len(fp.executed) != 0 { + t.Fatalf("Execute must not run for an undecodable body, got %v", fp.executed) + } +} diff --git a/internal/service/landscape.go b/internal/service/landscape.go index 7fcf066..6d2d66b 100644 --- a/internal/service/landscape.go +++ b/internal/service/landscape.go @@ -11,6 +11,7 @@ import ( "log/slog" "os" "path/filepath" + "runtime" "sort" "sync" "time" @@ -371,14 +372,23 @@ func (s *LandscapeService) Execute(ctx context.Context, runID uuid.UUID) error { if err := s.runs.SetRunStatus(ctx, runID, domain.LPRunRunning, nil); err != nil { return err } - // Re-read so started_at is available for result dataset names. + // Re-read so started_at is available for result dataset names; on failure + // keep the pre-transition copy and fall back to time.Now() for the names. if started, err := s.runs.GetRun(ctx, runID); err == nil { run = started + } else { + s.log.Warn("re-read landscape run after marking it running", "run_id", runID, "error", err) } state := &runState{produced: map[string]uuid.UUID{}} for _, stage := range landscape.Stages() { var g errgroup.Group + // Each node's Compute spins its own worker pool (up to + // min(GOMAXPROCS, windows, 8) workers, each holding a window buffer per + // input band), so peak memory is the product of the two limits. Bound + // the node fan-out here so that product stays sane: without a limit all + // four stage-0 nodes would run at once and multiply Compute's own cap. + g.SetLimit(maxConcurrentNodes()) for _, node := range stage { g.Go(func() error { return s.runNode(ctx, run, node, params, cat, state) }) } @@ -433,7 +443,8 @@ func (s *LandscapeService) runNode(ctx context.Context, run domain.LPRun, node l fp := fingerprint(node, refs, params) if err := s.runs.SetNodeStarted(ctx, run.ID, node.Code, fp, refs); err != nil { - return err + // Best-effort, so the node is not left pending while the run is failed. + return s.finishNodeFailed(ctx, run.ID, node, err) } if run.FromStage == nil || node.Stage < *run.FromStage { @@ -452,7 +463,11 @@ func (s *LandscapeService) runNode(ctx context.Context, run domain.LPRun, node l return s.finishNodeFailed(ctx, run.ID, node, err) } state.record(outputs) - return s.runs.SetNodeFinished(ctx, run.ID, node.Code, domain.LPNodeSucceeded, outputs, nil) + if err := s.runs.SetNodeFinished(ctx, run.ID, node.Code, domain.LPNodeSucceeded, outputs, nil); err != nil { + // Best-effort, so the node is not left running while the run is failed. + return s.finishNodeFailed(ctx, run.ID, node, err) + } + return nil } // finishNodeFailed records the failure on the node and returns the cause. @@ -678,6 +693,21 @@ func knownGroup(code string) bool { return ok } +// maxConcurrentNodes caps how many graph nodes of one stage run at the same +// time. Compute already parallelises within a node, so half the CPUs (at most +// 4, the widest stage) is enough to keep the machine busy without letting the +// two levels of parallelism multiply into an out-of-memory. +func maxConcurrentNodes() int { + n := runtime.GOMAXPROCS(0) / 2 + if n < 1 { + n = 1 + } + if n > 4 { + n = 4 + } + return n +} + // maxStage returns the highest stage number in the graph. func maxStage() int { highest := 0 diff --git a/internal/service/landscape_test.go b/internal/service/landscape_test.go index 10f04a1..94d34a0 100644 --- a/internal/service/landscape_test.go +++ b/internal/service/landscape_test.go @@ -356,6 +356,10 @@ func (j *stubLPJobs) EnqueueRun(_ context.Context, runID uuid.UUID) error { type stubComputeCall struct { node string inputs map[string]string + // bodies holds each input file's content at call time; seedDataset stores + // each object under its own storage key as its body, so a test can tell + // which dataset a node actually downloaded. + bodies map[string]string params landscape.Params } @@ -370,12 +374,19 @@ type stubCompute struct { func newStubCompute() *stubCompute { return &stubCompute{errs: map[string]error{}} } func (c *stubCompute) fn(_ context.Context, _ raster.IO, node landscape.Node, inputs map[string]string, outDir string, p landscape.Params) (map[string]string, error) { - c.mu.Lock() in := map[string]string{} + bodies := map[string]string{} for k, v := range inputs { in[k] = v + body, err := os.ReadFile(v) + if err != nil { + return nil, err + } + bodies[k] = string(body) } - c.calls = append(c.calls, stubComputeCall{node: node.Code, inputs: in, params: p}) + + c.mu.Lock() + c.calls = append(c.calls, stubComputeCall{node: node.Code, inputs: in, bodies: bodies, params: p}) err := c.errs[node.Code] c.mu.Unlock() if err != nil { @@ -512,7 +523,7 @@ func (f *lpFixture) seedDataset(t *testing.T, code, status string, fileType doma if err != nil { t.Fatalf("seed dataset %s: %v", code, err) } - if err := f.store.Put(ctx, key, bytes.NewReader([]byte("TIF")), 3, "image/tiff"); err != nil { + if err := f.store.Put(ctx, key, bytes.NewReader([]byte(key)), int64(len(key)), "image/tiff"); err != nil { t.Fatalf("seed object %s: %v", key, err) } return d @@ -1159,3 +1170,49 @@ func TestLandscapeService_Execute_NodeFailureFailsRunAndCleansUp(t *testing.T) { }) } } + +func TestLandscapeService_Execute_OverridePinsInputOverFresherDataset(t *testing.T) { + f := newLPFixture(t) + f.seedAllSlots(t) + + // Two ready rasters in the same slot group: the pinned one is the older of + // the two, so resolving by freshness alone would pick the other. + pinned := f.seedDataset(t, landscape.SlotZLCR, domain.DatasetStatusReady, domain.FileTypeRaster, "2023-01-31") + fresher := f.seedDataset(t, landscape.SlotZLCR, domain.DatasetStatusReady, domain.FileTypeRaster, "2026-01-31") + + run := f.runOnce(t, RunInput{Overrides: map[string]uuid.UUID{landscape.SlotZLCR: pinned.ID}}) + if run.Status != domain.LPRunSucceeded { + t.Fatalf("status = %s (%v), want succeeded", run.Status, run.Error) + } + + // Both Z nodes consume the slot, and both must record the pinned dataset. + for _, code := range []string{"00", "02"} { + got := refsByGroup(nodeByCode(t, run, code).Inputs)[landscape.SlotZLCR] + if got != pinned.ID { + if got == fresher.ID { + t.Fatalf("node %s resolved the fresher dataset %v, want the pinned %v", code, got, pinned.ID) + } + t.Fatalf("node %s input for %s = %v, want %v", code, landscape.SlotZLCR, got, pinned.ID) + } + } + + // The compute stub was handed the pinned object's bytes, not the fresher one's. + call, ok := f.compute.callFor("00") + if !ok { + t.Fatal("compute was not called for node 00") + } + path := call.inputs[landscape.SlotZLCR] + if path == "" { + t.Fatalf("node 00 got no path for %s (%v)", landscape.SlotZLCR, call.inputs) + } + if got := call.bodies[landscape.SlotZLCR]; got != pinned.StorageKey { + t.Fatalf("node 00 downloaded %q, want the pinned dataset's object %q", got, pinned.StorageKey) + } + + // A group with no override still resolves the freshest ready dataset. + newest := f.seedDataset(t, landscape.SlotUVeg, domain.DatasetStatusReady, domain.FileTypeRaster, "2026-02-28") + second := f.runOnce(t, RunInput{Overrides: map[string]uuid.UUID{landscape.SlotZLCR: pinned.ID}}) + if got := refsByGroup(nodeByCode(t, second, "01").Inputs)[landscape.SlotUVeg]; got != newest.ID { + t.Fatalf("node 01 input for %s = %v, want the freshest %v", landscape.SlotUVeg, got, newest.ID) + } +}