Bakhtiyar Issakhmetov
4051506b95
F2: Execute had no status guard, so a redelivered or stale message reran a finished run. Runs that are neither queued nor running are logged and acked without work; "running" still executes so a worker that crashed mid-run recovers. F4: failRun, finishNodeFailed and MarkRunFailed wrote the terminal status on the caller's context, which is exactly the context that is already cancelled when the run failed because of a shutdown or a sibling node — the write was refused and the run stayed stuck in "running". They now write on a context.WithoutCancel copy with a 10s timeout; the consumer's best-effort MarkRunFailed does the same. F5: peak memory scaled with the host's core count. LP_MAX_WORKERS (default 8) is now the total window-worker budget of a run and LP_MAX_NODES (default 2) the node fan-out; each node gets max(1, workers/nodes) through the new landscape.ComputeWithWorkers. Compute and service.ComputeFunc keep their signatures: Compute delegates with MaxWorkersPerNode, and a legacy ComputeFunc handed to NewLandscapeService is wrapped. F6: the stage loop used a plain errgroup, so a failing node let its siblings run to completion. It now uses errgroup.WithContext and passes the group context into runNode, cancelling siblings mid-compute. F7: wrap every float32 product in an explicit conversion so the compiler cannot fuse it into an FMA, keeping the kernels bit-identical to numpy's per-op rounding. F8: README's builder package list now matches the Dockerfile (adds g++), and the Landscape section documents the memory model and the two new knobs. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
95 lines
3.2 KiB
Go
95 lines
3.2 KiB
Go
package rabbitmq
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
const (
|
|
// LandscapeRunRoutingKey routes Landscape Program run jobs.
|
|
LandscapeRunRoutingKey = "landscape.run"
|
|
// LandscapeRunQueue receives Landscape Program run jobs.
|
|
LandscapeRunQueue = "gis.landscape.run"
|
|
// markRunFailedTimeout bounds the best-effort terminal status write made
|
|
// after Execute failed, on a context detached from the delivery's.
|
|
markRunFailedTimeout = 10 * time.Second
|
|
)
|
|
|
|
// LandscapeJob is the message body for a Landscape Program run job.
|
|
type LandscapeJob struct {
|
|
RunID uuid.UUID `json:"run_id"`
|
|
}
|
|
|
|
// LandscapeJobPublisher publishes Landscape Program run jobs; it implements
|
|
// service.LandscapeJobEnqueuer.
|
|
type LandscapeJobPublisher struct {
|
|
pub *Publisher
|
|
}
|
|
|
|
// NewLandscapeJobPublisher returns a LandscapeJobPublisher.
|
|
func NewLandscapeJobPublisher(pub *Publisher) *LandscapeJobPublisher {
|
|
return &LandscapeJobPublisher{pub: pub}
|
|
}
|
|
|
|
// EnqueueRun publishes an execution job for the given run.
|
|
func (p *LandscapeJobPublisher) EnqueueRun(ctx context.Context, runID uuid.UUID) error {
|
|
body, err := json.Marshal(LandscapeJob{RunID: runID})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return p.pub.Publish(ctx, LandscapeRunRoutingKey, body)
|
|
}
|
|
|
|
// LandscapeProcessor executes Landscape Program runs; implemented by the service.
|
|
type LandscapeProcessor interface {
|
|
Execute(ctx context.Context, runID uuid.UUID) error
|
|
// MarkRunFailed records a terminal failure on the run so it does not remain
|
|
// stuck in an in-progress status.
|
|
MarkRunFailed(ctx context.Context, runID uuid.UUID, reason string) error
|
|
}
|
|
|
|
// landscapeHandler executes one Landscape Program run per delivery.
|
|
type landscapeHandler struct {
|
|
proc LandscapeProcessor
|
|
log *slog.Logger
|
|
}
|
|
|
|
func (h landscapeHandler) Handle(ctx context.Context, d amqp.Delivery) error {
|
|
var job LandscapeJob
|
|
if err := json.Unmarshal(d.Body, &job); err != nil {
|
|
return fmt.Errorf("decode landscape run job: %w", err)
|
|
}
|
|
h.log.Info("processing landscape run", "run_id", job.RunID)
|
|
|
|
err := h.proc.Execute(ctx, job.RunID)
|
|
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. 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.
|
|
// Detached from ctx: Execute may well have failed *because* ctx was
|
|
// cancelled (shutdown), and the write must still land or the run stays
|
|
// "running" forever.
|
|
mctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), markRunFailedTimeout)
|
|
if markErr := h.proc.MarkRunFailed(mctx, job.RunID, err.Error()); markErr != nil {
|
|
h.log.Error("mark landscape run failed", "run_id", job.RunID, "error", markErr)
|
|
}
|
|
cancel()
|
|
}
|
|
return err
|
|
}
|
|
|
|
// NewLandscapeRunHandler returns a handler that executes Landscape Program runs.
|
|
func NewLandscapeRunHandler(p LandscapeProcessor, log *slog.Logger) Handler {
|
|
return landscapeHandler{proc: p, log: log}
|
|
}
|