fix(optimizer): bound queued solves to one deadline

.changeset/optimizer-deadline-budget.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Bound each optimizer request to one worker deadline so expired queued work no longer blocks newer plans or resets its budget between solver phases.

go/internal/mpc/optimizer_transport.go

@@ -12,7 +12,6 @@ import (
 	"os/exec"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/srcfl/ftw/go/internal/optimizercontract"
@@ -86,7 +85,7 @@ type ProcessTransportConfig struct {
 type ProcessTransport struct {
 	cfg ProcessTransportConfig
 
-	mu        sync.Mutex
+	mu        *contextGate
 	cmd       *exec.Cmd
 	stdin     io.WriteCloser
 	scanner   *bufio.Scanner
@@ -98,16 +97,22 @@ func NewProcessTransport(cfg ProcessTransportConfig) (*ProcessTransport, error)
 	if len(cfg.Command) == 0 || strings.TrimSpace(cfg.Command[0]) == "" {
 		return nil, errors.New("optimizer command is empty")
 	}
-	return &ProcessTransport{cfg: cfg}, nil
+	return &ProcessTransport{cfg: cfg, mu: newContextGate()}, nil
 }
 
 func (t *ProcessTransport) RoundTrip(ctx context.Context, payload []byte) ([]byte, error) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
+	if err := t.mu.acquire(ctx); err != nil {
+		return nil, err
+	}
+	defer t.mu.release()
 	t.cancelIdleStopLocked()
 	if err := t.ensureStartedLocked(); err != nil {
 		return nil, err
 	}
+	if err := ctx.Err(); err != nil {
+		t.scheduleIdleStopLocked()
+		return nil, err
+	}
 	if _, err := t.stdin.Write(append(append([]byte(nil), payload...), '\n')); err != nil {
 		t.stopLocked()
 		return nil, fmt.Errorf("write optimizer request: %w", err)
@@ -122,15 +127,18 @@ func (t *ProcessTransport) RoundTrip(ctx context.Context, payload []byte) ([]byt
 }
 
 func (t *ProcessTransport) Health(ctx context.Context) (OptimizerRuntimeInfo, error) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	t.cancelIdleStopLocked()
-	if err := ctx.Err(); err != nil {
+	if err := t.mu.acquire(ctx); err != nil {
 		return OptimizerRuntimeInfo{}, err
 	}
+	defer t.mu.release()
+	t.cancelIdleStopLocked()
 	if err := t.ensureStartedLocked(); err != nil {
 		return OptimizerRuntimeInfo{}, err
 	}
+	if err := ctx.Err(); err != nil {
+		t.scheduleIdleStopLocked()
+		return OptimizerRuntimeInfo{}, err
+	}
 	payload, _ := json.Marshal(map[string]any{
 		"type":             "handshake",
 		"protocol_version": OptimizerProtocolVersion,
@@ -216,8 +224,8 @@ func (t *ProcessTransport) scheduleIdleStopLocked() {
 	t.cancelIdleStopLocked()
 	var timer *time.Timer
 	timer = time.AfterFunc(t.cfg.IdleTimeout, func() {
-		t.mu.Lock()
-		defer t.mu.Unlock()
+		_ = t.mu.acquire(context.Background())
+		defer t.mu.release()
 		if t.idleTimer != timer {
 			return
 		}
@@ -239,8 +247,8 @@ func (t *ProcessTransport) stopLocked() {
 }
 
 func (t *ProcessTransport) Close() error {
-	t.mu.Lock()
-	defer t.mu.Unlock()
+	_ = t.mu.acquire(context.Background())
+	defer t.mu.release()
 	t.cancelIdleStopLocked()
 	if t.cmd == nil {
 		return nil
@@ -256,6 +264,47 @@ func (t *ProcessTransport) Close() error {
 	}
 }
 
+// contextGate serializes access to the warm process without hiding queue wait
+// from the caller's deadline. A request canceled while it waits never reaches
+// stdin, so it cannot occupy the worker after its result has become useless.
+type contextGate struct {
+	token chan struct{}
+}
+
+func newContextGate() *contextGate {
+	g := &contextGate{token: make(chan struct{}, 1)}
+	g.token <- struct{}{}
+	return g
+}
+
+func (g *contextGate) acquire(ctx context.Context) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-g.token:
+		if err := ctx.Err(); err != nil {
+			g.release()
+			return err
+		}
+		return nil
+	}
+}
+
+func (g *contextGate) release() {
+	g.token <- struct{}{}
+}
+
+func (g *contextGate) Lock() {
+	_ = g.acquire(context.Background())
+}
+
+func (g *contextGate) Unlock() {
+	g.release()
+}
+
 type UnixTransport struct{ socketPath string }
 
 func NewUnixTransport(socketPath string) *UnixTransport {

go/internal/mpc/optimizer_transport_test.go

@@ -37,6 +37,56 @@ func TestOptimizerProtocolVersionKeepsContractAlias(t *testing.T) {
 	}
 }
 
+func TestContextGateDropsCanceledWaiter(t *testing.T) {
+	gate := newContextGate()
+	if err := gate.acquire(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	waiting := make(chan struct{})
+	errCh := make(chan error, 1)
+	go func() {
+		close(waiting)
+		errCh <- gate.acquire(ctx)
+	}()
+	<-waiting
+	cancel()
+
+	select {
+	case err := <-errCh:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("acquire error = %v, want context.Canceled", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("canceled waiter remained blocked")
+	}
+
+	gate.release()
+	acquireCtx, acquireCancel := context.WithTimeout(context.Background(), time.Second)
+	defer acquireCancel()
+	if err := gate.acquire(acquireCtx); err != nil {
+		t.Fatalf("gate stayed occupied after canceled waiter: %v", err)
+	}
+	gate.release()
+}
+
+func TestProcessTransportRejectsCanceledContextBeforeWorkerLookup(t *testing.T) {
+	transport, err := NewProcessTransport(ProcessTransportConfig{
+		Command: []string{"ftw-worker-that-does-not-exist"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err = transport.RoundTrip(ctx, []byte(`{}`))
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("RoundTrip error = %v, want context.Canceled", err)
+	}
+}
+
 func TestUnixTransportHandshakeAndRoundTrip(t *testing.T) {
 	path := fmt.Sprintf("/tmp/ftw-opt-%d.sock", time.Now().UnixNano())
 	t.Cleanup(func() { _ = os.Remove(path) })

optimizer/ftw_optimizer/deadline.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+from .protocol import positive_number, require_dict
+
+
+class SolveDeadlineExceeded(RuntimeError):
+    """The request's one worker-side time budget has been spent."""
+
+
+@dataclass(frozen=True)
+class SolveDeadline:
+    expires_at: float
+    clock: Callable[[], float] = field(
+        default=time.perf_counter,
+        repr=False,
+        compare=False,
+    )
+
+    @classmethod
+    def from_payload(
+        cls,
+        payload: dict[str, Any],
+        *,
+        started_at: float | None = None,
+        clock: Callable[[], float] = time.perf_counter,
+    ) -> SolveDeadline:
+        settings = require_dict(payload.get("settings", {}), "settings")
+        budget_s = positive_number(
+            settings.get("time_limit_s", 2.0),
+            "settings.time_limit_s",
+        )
+        if started_at is None:
+            started_at = clock()
+        return cls(started_at + budget_s, clock)
+
+    def remaining_s(self, phase: str = "optimizer request") -> float:
+        remaining = self.expires_at - self.clock()
+        if remaining <= 0.0:
+            raise SolveDeadlineExceeded(f"{phase} deadline exceeded")
+        return remaining
+
+    def check(self, phase: str = "optimizer request") -> None:
+        self.remaining_s(phase)

optimizer/ftw_optimizer/direct_highs.py

@@ -9,6 +9,7 @@
 import numpy as np
 
 from . import SCHEMA_VERSION
+from .deadline import SolveDeadline, SolveDeadlineExceeded
 from .model import (
     _arbitrage_spread_ore_kwh,
     _solver_options,
@@ -160,7 +161,7 @@ def solve_direct_highs(
     *,
     shared: bool = False,
     exact_shared_baseline: bool = False,
-    deadline: float | None = None,
+    deadline: SolveDeadline | float | None = None,
     prior_build_ms: float = 0.0,
     prior_solver_ms: float = 0.0,
 ) -> dict[str, Any]:
@@ -171,9 +172,11 @@ def solve_direct_highs(
     if prepared.discrete or prepared.unsafe_cycle or prepared.unsafe_meter_split:
         raise DirectHighsError("direct HiGHS path requires a cycle-safe continuous tariff")
     if deadline is None:
-        deadline = started + float(
-            _solver_options(prepared.settings, "HIGHS")["time_limit"]
+        deadline = SolveDeadline(
+            started
+            + float(_solver_options(prepared.settings, "HIGHS")["time_limit"])
         )
+    _remaining_time_s(deadline)
     build_started = time.perf_counter()
     model = SparseModel()
     m = len(prepared.scenario_set.scenarios)
@@ -602,8 +605,12 @@ def solve_direct_highs(
         time_limit_s=_remaining_time_s(deadline),
     )
     build_ms = (time.perf_counter() - build_started) * 1000.0
+    _require_ok(
+        highs.setOptionValue("time_limit", _remaining_time_s(deadline)),
+        "set service time limit",
+    )
     solver_started = time.perf_counter()
-    _run_optimal(highs, "service")
+    _run_optimal(highs, "service", deadline)
     best_service = max(0.0, float(highs.getObjectiveValue()))
     _require_ok(
         highs.changeRowsBounds(
@@ -623,7 +630,7 @@ def solve_direct_highs(
         highs.setOptionValue("time_limit", _remaining_time_s(deadline)),
         "set economic time limit",
     )
-    _run_optimal(highs, "economic")
+    _run_optimal(highs, "economic", deadline)
     solver_ms = (time.perf_counter() - solver_started) * 1000.0
     mip_gap = float(highs.getInfo().mip_gap) if model.integer else None
     solution = np.asarray(highs.getSolution().col_value, dtype=np.float64)
@@ -857,10 +864,12 @@ def _add(coefficients: dict[int, float], index: int, value: float) -> None:
     coefficients[index] = coefficients.get(index, 0.0) + value
 
 
-def _remaining_time_s(deadline: float) -> float:
+def _remaining_time_s(deadline: SolveDeadline | float) -> float:
+    if isinstance(deadline, SolveDeadline):
+        return deadline.remaining_s("direct HiGHS solve")
     remaining = deadline - time.perf_counter()
     if remaining <= 0.0:
-        raise DirectHighsError("direct HiGHS time budget exhausted")
+        raise SolveDeadlineExceeded("direct HiGHS solve deadline exceeded")
     return remaining
 
 
@@ -910,8 +919,18 @@ def _require_ok(status: highspy.HighsStatus, operation: str) -> None:
         raise DirectHighsError(f"HiGHS failed to {operation}: {status}")
 
 
-def _run_optimal(highs: highspy.Highs, phase: str) -> None:
-    _require_ok(highs.run(), f"run {phase} solve")
+def _run_optimal(
+    highs: highspy.Highs,
+    phase: str,
+    deadline: SolveDeadline | float,
+) -> None:
+    run_status = highs.run()
     status = highs.getModelStatus()
+    if status == highspy.HighsModelStatus.kTimeLimit:
+        raise SolveDeadlineExceeded(
+            f"direct HiGHS {phase} solve deadline exceeded"
+        )
+    _require_ok(run_status, f"run {phase} solve")
     if status != highspy.HighsModelStatus.kOptimal:
         raise DirectHighsError(f"HiGHS {phase} solve failed with status {status}")
+    _remaining_time_s(deadline)