infureal/gis
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(landscape): bound node fan-out, cover override resolution, test the run consumer

Browse Source 
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Bakhtiyar Issakhmetov
2026-09-05 13:49:27 +05:00
parent 5fba77b59c
commit d74277552f
4 changed files with 188 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
internal/service/landscape.go
View File

@@ -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