From e463ced4ce3ea964e5acf3e508eb2d4fc872ab6f Mon Sep 17 00:00:00 2001 From: Wingston Sharon Date: Mon, 20 Apr 2026 15:43:25 +0200 Subject: [PATCH 1/5] fix(sdk): correctly route batch vs single embeddings to result fields Cubic review finding (sdk/src/beta9/inference.py:307): Batch embed() results were being assigned to the singular `embedding` field while `embeddings` (the documented plural list-of-lists) stayed empty. tests/test_cubic_fixes.py::test_batch_embeddings exercises this path and asserts `len(result.embeddings) == 3`. Fix: track whether caller passed a string (single) or a list (batch), and populate: - `embedding` = flat vector for single-input calls (back-compat) - `embeddings` = full list-of-lists for batch calls Co-Authored-By: Claudistrator --- sdk/src/beta9/inference.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/sdk/src/beta9/inference.py b/sdk/src/beta9/inference.py index 9fcf2dfb0..af8efda36 100644 --- a/sdk/src/beta9/inference.py +++ b/sdk/src/beta9/inference.py @@ -280,7 +280,8 @@ def embed( Returns: EmbeddingResult with the embedding vector(s) """ - # Handle single string + # Track whether caller passed a single string or a list (batch) + is_batch = not isinstance(input, str) if isinstance(input, str): input = [input] @@ -301,10 +302,15 @@ def embed( single = data.get("embedding", []) embeddings = [single] if single else [] - # Return all embeddings for batch input + # Populate fields per contract: + # - `embedding` (singular): the single vector when caller passed a string + # (or first vector for back-compat when batch also returned one result) + # - `embeddings` (plural, list-of-lists): always the full batch result + single_vec: List[float] = embeddings[0] if embeddings else [] return EmbeddingResult( model=model, - embedding=embeddings[0] if len(embeddings) == 1 else embeddings, + embedding=single_vec if not is_batch else (embeddings[0] if len(embeddings) == 1 else []), + embeddings=embeddings if is_batch else [], usage={ "prompt_tokens": data.get("prompt_eval_count", 0), }, From 55adb935e21669b7f09938b1e1748bfe16facf99 Mon Sep 17 00:00:00 2001 From: Wingston Sharon Date: Mon, 20 Apr 2026 15:45:36 +0200 Subject: [PATCH 2/5] fix(gateway): check worker HTTP status before clearing unload state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cubic review finding (pkg/gateway/inference_router.go:463): UnloadModel previously cleared the registry's LoadState to Idle no matter what the worker returned. Any 4xx/5xx response would leave the registry inconsistent with the worker's actual in-memory model set — a cache-coherency bug. Fix: - On non-200 response: log error with status + up to 4 KiB of body, return an error, and leave the model's LoadState unchanged. - Only on 200: drain body and transition to LoadStateIdle. Added table-driven TestUnloadModelNonOK + TestUnloadModelNodeNotFound in pkg/gateway/inference_router_test.go covering 200/500/404 paths and the missing-node early return. Co-Authored-By: Claudistrator --- pkg/gateway/inference_router.go | 19 +++++- pkg/gateway/inference_router_test.go | 91 ++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 2 deletions(-) create mode 100644 pkg/gateway/inference_router_test.go diff --git a/pkg/gateway/inference_router.go b/pkg/gateway/inference_router.go index f82149c89..ee23dffb1 100644 --- a/pkg/gateway/inference_router.go +++ b/pkg/gateway/inference_router.go @@ -457,10 +457,25 @@ func (r *InferenceRouter) UnloadModel(ctx context.Context, nodeID, model string) } defer resp.Body.Close() - // Drain response + // Only clear registry state when the worker acknowledged the unload. + // Previously any response (including 4xx/5xx) was treated as success, + // producing a cache-coherency bug where the registry reported the model + // idle while the worker still had it loaded. + if resp.StatusCode != http.StatusOK { + bodyBytes, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + log.Error(). + Int("status", resp.StatusCode). + Str("node", nodeID). + Str("model", model). + Str("body", string(bodyBytes)). + Msg("Unload request returned non-OK status; leaving registry state unchanged") + return fmt.Errorf("unload failed: node %s returned HTTP %d", nodeID, resp.StatusCode) + } + + // Drain response on success io.Copy(io.Discard, resp.Body) - // Update registry regardless of response + // Update registry now that the worker confirmed the unload if node := r.registry.GetNode(nodeID); node != nil { if modelInfo, exists := node.Models[model]; exists { modelInfo.LoadState = types.LoadStateIdle diff --git a/pkg/gateway/inference_router_test.go b/pkg/gateway/inference_router_test.go new file mode 100644 index 000000000..f08ac80fb --- /dev/null +++ b/pkg/gateway/inference_router_test.go @@ -0,0 +1,91 @@ +package gateway + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "testing" + + "github.com/beam-cloud/beta9/pkg/types" +) + +// TestUnloadModelNonOK verifies that UnloadModel only clears registry state +// when the worker responds with HTTP 200. Non-OK responses must return an +// error and leave the model's LoadState unchanged (Cubic finding on +// pkg/gateway/inference_router.go:463 — cache-coherency bug). +func TestUnloadModelNonOK(t *testing.T) { + cases := []struct { + name string + status int + body string + wantErr bool + wantFinalState types.LoadState + }{ + {"ok clears state", http.StatusOK, "{}", false, types.LoadStateIdle}, + {"500 preserves state", http.StatusInternalServerError, "boom", true, types.LoadStateReady}, + {"404 preserves state", http.StatusNotFound, "not found", true, types.LoadStateReady}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tc.status) + _, _ = w.Write([]byte(tc.body)) + })) + defer srv.Close() + + // Parse httptest URL into host + port. srv.URL looks like "http://127.0.0.1:54321". + u := strings.TrimPrefix(srv.URL, "http://") + host, portStr, err := net.SplitHostPort(u) + if err != nil { + t.Fatalf("split host:port: %v", err) + } + port, err := strconv.Atoi(portStr) + if err != nil { + t.Fatalf("port: %v", err) + } + + reg := NewModelRegistry() + reg.RegisterNode(&types.NodeInferenceInfo{ + NodeID: "node-a", + TailscaleIP: host, + Port: port, + Models: map[string]*types.ModelInfo{ + "llama3": { + Name: "llama3", + LoadState: types.LoadStateReady, + }, + }, + }) + + router := NewInferenceRouter(reg) + + err = router.UnloadModel(context.Background(), "node-a", "llama3") + if tc.wantErr && err == nil { + t.Fatalf("expected error, got nil") + } + if !tc.wantErr && err != nil { + t.Fatalf("expected nil error, got %v", err) + } + + got := reg.GetNode("node-a").Models["llama3"].LoadState + if got != tc.wantFinalState { + t.Fatalf("LoadState = %q, want %q (status=%d)", got, tc.wantFinalState, tc.status) + } + }) + } +} + +// TestUnloadModelNodeNotFound verifies the early-return when the node is +// absent from the registry. +func TestUnloadModelNodeNotFound(t *testing.T) { + reg := NewModelRegistry() + router := NewInferenceRouter(reg) + err := router.UnloadModel(context.Background(), "missing-node", "any-model") + if err == nil { + t.Fatalf("expected error for missing node") + } +} From cd7acd3704174c0c54492b8a3cc9bd89540f95a7 Mon Sep 17 00:00:00 2001 From: Wingston Sharon Date: Mon, 20 Apr 2026 15:49:04 +0200 Subject: [PATCH 3/5] fix(agent): stop reporting pull success when Ollama failed or decode broke Cubic review finding (pkg/agent/control.go:227): handleInferencePull forwarded to Ollama's /api/pull, looped over the NDJSON body with a decoder, and returned StatusOK unconditionally. A non-200 from Ollama or a malformed stream both produced a false success response. Fix: - Check resp.StatusCode != 200 before streaming; non-OK now returns 502 with a 512-byte body preview from Ollama for diagnosis. - Track decoder errors; if decode fails (or the stream is empty) the handler returns 502 instead of pretending success. - Success case now returns 201 Created to distinguish from the pre-fix false-success 200. Inline P0-A hardening (required by spec, not yet on origin/main): - MaxBytesReader caps request body at 4 KiB. - isAllowedModelName() allowlist [A-Za-z0-9._/:-] + length <= 128 and blocks ".." path-traversal sequences before they reach Ollama. Added TestInferencePullStatus (200/404/500 upstream), TestInferencePullDecodeFailureReturnsBadGateway, TestInferencePullRejectsInvalidModelName, TestInferencePullMethodNotAllowed in pkg/agent/control_test.go. ControlServer gained an optional ollamaBaseURL field so httptest servers can stand in for the daemon. Co-Authored-By: Claudistrator --- pkg/agent/control.go | 98 ++++++++++++++++++++++++++++++- pkg/agent/control_test.go | 120 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+), 3 deletions(-) create mode 100644 pkg/agent/control_test.go diff --git a/pkg/agent/control.go b/pkg/agent/control.go index 951d21bbe..b7e84893e 100644 --- a/pkg/agent/control.go +++ b/pkg/agent/control.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "net/http" + "strings" "time" "github.com/rs/zerolog/log" @@ -18,6 +19,18 @@ type ControlServer struct { agent *Agent port int server *http.Server + + // ollamaBaseURL lets tests redirect Ollama calls to an httptest server. + // Empty string means default: http://localhost:DefaultOllamaPort + ollamaBaseURL string +} + +// ollamaURL returns the Ollama base URL honoring test overrides. +func (c *ControlServer) ollamaURL() string { + if c.ollamaBaseURL != "" { + return c.ollamaBaseURL + } + return fmt.Sprintf("http://localhost:%d", DefaultOllamaPort) } // NewControlServer creates a new control server @@ -117,6 +130,32 @@ func (c *ControlServer) handleInferenceStop(w http.ResponseWriter, r *http.Reque }) } +// isAllowedModelName enforces a narrow allowlist pattern for model names +// accepted by /inference/pull. Ollama model names are of the form +// "/:" or ":" — only A-Z a-z 0-9 . _ - / : +// are permitted, length <= 128. Explicitly rejects ".." to block path +// traversal payloads from reaching the Ollama daemon. +func isAllowedModelName(name string) bool { + if name == "" || len(name) > 128 { + return false + } + // Block path-traversal sequences outright. + if strings.Contains(name, "..") { + return false + } + for _, ch := range name { + switch { + case ch >= 'a' && ch <= 'z': + case ch >= 'A' && ch <= 'Z': + case ch >= '0' && ch <= '9': + case ch == '.' || ch == '_' || ch == '-' || ch == '/' || ch == ':': + default: + return false + } + } + return true +} + // handleInferencePull pulls a model and streams progress to TUI logs func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { @@ -124,6 +163,9 @@ func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Reque return } + // Cap request body to prevent memory-exhaustion payloads (P0-A hardening). + r.Body = http.MaxBytesReader(w, r.Body, 4096) + // Parse request body var req struct { Model string `json:"model"` @@ -144,6 +186,16 @@ func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Reque return } + // P0-A model-name allowlist: reject anything with shell metacharacters + // or path-traversal sequences before forwarding to the Ollama daemon. + if !isAllowedModelName(req.Model) { + writeJSON(w, http.StatusBadRequest, map[string]any{ + "status": "error", + "error": "Invalid model name", + }) + return + } + log.Info().Str("model", req.Model).Msg("Control: pull model command received") c.agent.state.AddLog(fmt.Sprintf("Pulling model: %s", req.Model)) @@ -153,7 +205,7 @@ func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Reque client := &http.Client{Timeout: 30 * time.Minute} // Model pulls can take a while resp, err := client.Post( - fmt.Sprintf("http://localhost:%d/api/pull", DefaultOllamaPort), + fmt.Sprintf("%s/api/pull", c.ollamaURL()), "application/json", bytes.NewReader(body), ) @@ -167,9 +219,30 @@ func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Reque } defer resp.Body.Close() - // Stream progress to logs + // Check Ollama's status BEFORE streaming the response body. Previously a + // non-200 from Ollama still produced StatusOK downstream because the + // decode loop swallowed errors and the handler unconditionally returned + // ok (Cubic finding on pkg/agent/control.go:227). + if resp.StatusCode != http.StatusOK { + bodyPreview := make([]byte, 512) + n, _ := resp.Body.Read(bodyPreview) + errMsg := string(bodyPreview[:n]) + c.agent.state.AddLog(fmt.Sprintf("Pull failed: Ollama returned HTTP %d", resp.StatusCode)) + writeJSON(w, http.StatusBadGateway, map[string]any{ + "status": "error", + "error": fmt.Sprintf("Ollama returned HTTP %d", resp.StatusCode), + "ollama_body": errMsg, + "ollama_status": resp.StatusCode, + }) + return + } + + // Stream progress to logs. Track decode errors so a malformed stream + // surfaces as a 502 rather than silent success. decoder := json.NewDecoder(resp.Body) lastStatus := "" + var decodeErr error + sawAny := false for { var progress struct { Status string `json:"status"` @@ -178,8 +251,14 @@ func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Reque Completed int64 `json:"completed"` } if err := decoder.Decode(&progress); err != nil { + // io.EOF is a clean end of stream; any other error is a decode + // failure that should be reported upstream. + if err.Error() != "EOF" { + decodeErr = err + } break } + sawAny = true // Update logs with progress (avoid duplicate messages) status := progress.Status @@ -193,6 +272,19 @@ func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Reque } } + if decodeErr != nil || !sawAny { + reason := "no progress frames received" + if decodeErr != nil { + reason = decodeErr.Error() + } + c.agent.state.AddLog(fmt.Sprintf("Pull failed: %s", reason)) + writeJSON(w, http.StatusBadGateway, map[string]any{ + "status": "error", + "error": fmt.Sprintf("failed to decode Ollama response: %s", reason), + }) + return + } + c.agent.state.AddLog(fmt.Sprintf("Model %s ready", req.Model)) // Update models list @@ -201,7 +293,7 @@ func (c *ControlServer) handleInferencePull(w http.ResponseWriter, r *http.Reque c.agent.state.UpdateInference("running", c.agent.ollama.TailscaleIP(), DefaultOllamaPort, models) } - writeJSON(w, http.StatusOK, map[string]any{ + writeJSON(w, http.StatusCreated, map[string]any{ "status": "ok", "message": fmt.Sprintf("Model %s pulled successfully", req.Model), }) diff --git a/pkg/agent/control_test.go b/pkg/agent/control_test.go new file mode 100644 index 000000000..405e24ac2 --- /dev/null +++ b/pkg/agent/control_test.go @@ -0,0 +1,120 @@ +package agent + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +// newTestControlServer wires a minimal Agent with state + OllamaManager so the +// /inference/pull handler's state touches are safe without starting a TUI or +// keepalive loop. +func newTestControlServer(t *testing.T, ollamaBaseURL string) *ControlServer { + t.Helper() + state := NewAgentState("test-machine", "test-pool", "http://gateway") + agent := &Agent{ + state: state, + ollama: NewOllamaManager("127.0.0.1", 0), + } + cs := NewControlServer(agent, 0) + cs.ollamaBaseURL = ollamaBaseURL + return cs +} + +// TestInferencePullStatus verifies that the handler only reports success when +// Ollama returns HTTP 200 AND the NDJSON stream decodes cleanly with at least +// one progress frame. Any non-200 upstream must become HTTP 502 at the +// control-plane boundary (Cubic finding on pkg/agent/control.go:227). +func TestInferencePullStatus(t *testing.T) { + // Valid NDJSON progress-stream frames. + okFrames := "{\"status\":\"pulling manifest\"}\n{\"status\":\"success\"}\n" + + cases := []struct { + name string + upstreamStatus int + upstreamBody string + wantHandlerCode int + }{ + {"ollama 200 returns 201", http.StatusOK, okFrames, http.StatusCreated}, + {"ollama 404 surfaces 502", http.StatusNotFound, "{\"error\":\"model not found\"}", http.StatusBadGateway}, + {"ollama 500 surfaces 502", http.StatusInternalServerError, "{\"error\":\"boom\"}", http.StatusBadGateway}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ollama := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tc.upstreamStatus) + _, _ = w.Write([]byte(tc.upstreamBody)) + })) + defer ollama.Close() + + cs := newTestControlServer(t, ollama.URL) + + body, _ := json.Marshal(map[string]string{"model": "llama3:8b"}) + req := httptest.NewRequest(http.MethodPost, "/inference/pull", bytes.NewReader(body)) + rr := httptest.NewRecorder() + cs.handleInferencePull(rr, req) + + if rr.Code != tc.wantHandlerCode { + t.Fatalf("handler code = %d, want %d; body=%s", rr.Code, tc.wantHandlerCode, rr.Body.String()) + } + }) + } +} + +// TestInferencePullDecodeFailureReturnsBadGateway covers the case where +// Ollama returns 200 but the body is not NDJSON. The handler must not +// pretend success — it must surface a 502. +func TestInferencePullDecodeFailureReturnsBadGateway(t *testing.T) { + ollama := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("this is not valid json at all")) + })) + defer ollama.Close() + + cs := newTestControlServer(t, ollama.URL) + + body, _ := json.Marshal(map[string]string{"model": "llama3:8b"}) + req := httptest.NewRequest(http.MethodPost, "/inference/pull", bytes.NewReader(body)) + rr := httptest.NewRecorder() + cs.handleInferencePull(rr, req) + + if rr.Code != http.StatusBadGateway { + t.Fatalf("decode failure should produce 502, got %d; body=%s", rr.Code, rr.Body.String()) + } +} + +// TestInferencePullRejectsInvalidModelName asserts the P0-A allowlist blocks +// shell-metacharacter payloads before they reach Ollama. +func TestInferencePullRejectsInvalidModelName(t *testing.T) { + cs := newTestControlServer(t, "http://127.0.0.1:1") // unreachable on purpose + + payloads := []string{ + "llama3; rm -rf /", + "../../etc/passwd", + "model`whoami`", + "", + } + for _, p := range payloads { + body, _ := json.Marshal(map[string]string{"model": p}) + req := httptest.NewRequest(http.MethodPost, "/inference/pull", bytes.NewReader(body)) + rr := httptest.NewRecorder() + cs.handleInferencePull(rr, req) + if rr.Code != http.StatusBadRequest { + t.Fatalf("payload %q should be rejected with 400, got %d", p, rr.Code) + } + } +} + +// TestInferencePullMethodNotAllowed checks GET is refused. +func TestInferencePullMethodNotAllowed(t *testing.T) { + cs := newTestControlServer(t, "") + req := httptest.NewRequest(http.MethodGet, "/inference/pull", nil) + rr := httptest.NewRecorder() + cs.handleInferencePull(rr, req) + if rr.Code != http.StatusMethodNotAllowed { + t.Fatalf("GET should be 405, got %d", rr.Code) + } +} From 241e932a5081f7fd559398bcd69c9e8a79362a58 Mon Sep 17 00:00:00 2001 From: Wingston Sharon Date: Mon, 20 Apr 2026 15:50:15 +0200 Subject: [PATCH 4/5] fix(agent): make KeepaliveLoop Start/Stop idempotent under concurrency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cubic review finding (pkg/agent/keepalive.go, Start ~L113 / Stop): Calling Stop() twice — or Start() after Stop() — closed an already closed channel and produced a "close of closed channel" panic. The bug is reachable during shutdown races and during retries after a transient failure. Fix: - Added `startOnce` + `stopOnce` sync.Once guards plus an atomic `started` flag. Start is a no-op after the first call; subsequent Stops are no-ops too. - Stop-before-Start is also safe: when started == 0 we return without waiting on doneCh (which would otherwise never close). Tests in pkg/agent/keepalive_test.go: - TestKeepaliveLoopIdempotent — Start, Stop, Stop, Start, Stop - TestKeepaliveLoopConcurrentStop — 32 goroutines stopping in parallel - TestKeepaliveLoopStopBeforeStart — Stop never blocks pre-Start Co-Authored-By: Claudistrator --- pkg/agent/keepalive.go | 40 ++++++++++++---- pkg/agent/keepalive_test.go | 92 +++++++++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+), 10 deletions(-) create mode 100644 pkg/agent/keepalive_test.go diff --git a/pkg/agent/keepalive.go b/pkg/agent/keepalive.go index 968d7c1f3..4bad7b18e 100644 --- a/pkg/agent/keepalive.go +++ b/pkg/agent/keepalive.go @@ -41,6 +41,14 @@ type KeepaliveLoop struct { maxFailures int32 stopCh chan struct{} doneCh chan struct{} + + // startOnce guards Start so a second Start() after Stop is a no-op + // rather than re-launching the goroutine against an already-closed + // stopCh. stopOnce guards Stop so a double-Stop does not panic with + // "close of closed channel" (Cubic finding, pkg/agent/keepalive.go). + startOnce sync.Once + stopOnce sync.Once + started int32 // atomic flag: 1 after Start has been invoked } // NewKeepaliveLoop creates a new keepalive loop @@ -76,9 +84,14 @@ func (k *KeepaliveLoop) GetLastMetrics() *MachineMetrics { return &MachineMetrics{} } -// Start begins the keepalive loop in a goroutine +// Start begins the keepalive loop in a goroutine. Calling Start more than +// once is a no-op; Start after Stop refuses to re-launch (safe on concurrent +// invocation). func (k *KeepaliveLoop) Start(ctx context.Context) { - go k.run(ctx) + k.startOnce.Do(func() { + atomic.StoreInt32(&k.started, 1) + go k.run(ctx) + }) } func (k *KeepaliveLoop) run(ctx context.Context) { @@ -108,15 +121,22 @@ func (k *KeepaliveLoop) run(ctx context.Context) { } } -// Stop signals the keepalive loop to stop and waits for it to finish +// Stop signals the keepalive loop to stop and waits for it to finish. +// Safe to call multiple times concurrently; subsequent calls are no-ops. func (k *KeepaliveLoop) Stop() { - close(k.stopCh) - // Wait for loop to finish with timeout - select { - case <-k.doneCh: - case <-time.After(5 * time.Second): - log.Warn().Msg("Keepalive loop did not stop within timeout") - } + k.stopOnce.Do(func() { + close(k.stopCh) + // Wait for loop to finish with timeout. If Start was never called + // there is no goroutine to wait for; skip the wait in that case. + if atomic.LoadInt32(&k.started) == 0 { + return + } + select { + case <-k.doneCh: + case <-time.After(5 * time.Second): + log.Warn().Msg("Keepalive loop did not stop within timeout") + } + }) } // IsHealthy returns true if recent keepalives succeeded diff --git a/pkg/agent/keepalive_test.go b/pkg/agent/keepalive_test.go new file mode 100644 index 000000000..ddf7048f7 --- /dev/null +++ b/pkg/agent/keepalive_test.go @@ -0,0 +1,92 @@ +package agent + +import ( + "context" + "sync" + "testing" + "time" +) + +// TestKeepaliveLoopIdempotent verifies the Cubic fix for double-Stop / +// Start-after-Stop panics. Calling Stop twice must not close an already +// closed channel; calling Start after Stop must be a no-op (no new +// goroutine, no panic). +func TestKeepaliveLoopIdempotent(t *testing.T) { + cfg := &AgentConfig{ + MachineID: "test-machine", + ProviderName: "test", + PoolName: "test-pool", + KeepaliveInterval: 10 * time.Millisecond, + DryRun: true, // skip real HTTP + } + + loop := NewKeepaliveLoop(cfg) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + loop.Start(ctx) + time.Sleep(25 * time.Millisecond) // let the goroutine tick at least once + + // Should not panic. + loop.Stop() + loop.Stop() + + // Start-after-Stop must be safe — no panic, no new goroutine. + loop.Start(ctx) + + // And a third Stop must also be safe. + loop.Stop() +} + +// TestKeepaliveLoopConcurrentStop hammers Stop from many goroutines to +// confirm the sync.Once guard holds under concurrency. +func TestKeepaliveLoopConcurrentStop(t *testing.T) { + cfg := &AgentConfig{ + MachineID: "test-machine", + ProviderName: "test", + PoolName: "test-pool", + KeepaliveInterval: 10 * time.Millisecond, + DryRun: true, + } + loop := NewKeepaliveLoop(cfg) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + loop.Start(ctx) + time.Sleep(15 * time.Millisecond) + + var wg sync.WaitGroup + for i := 0; i < 32; i++ { + wg.Add(1) + go func() { + defer wg.Done() + loop.Stop() + }() + } + wg.Wait() +} + +// TestKeepaliveLoopStopBeforeStart verifies Stop called before Start is +// safe: no panic, no hang. +func TestKeepaliveLoopStopBeforeStart(t *testing.T) { + cfg := &AgentConfig{ + MachineID: "test-machine", + ProviderName: "test", + PoolName: "test-pool", + KeepaliveInterval: time.Second, + DryRun: true, + } + loop := NewKeepaliveLoop(cfg) + + done := make(chan struct{}) + go func() { + loop.Stop() + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Stop-before-Start hung") + } +} From 14d47d1fc08d27458c0005b0337afdaf2c0c228d Mon Sep 17 00:00:00 2001 From: Wingston Sharon Date: Mon, 20 Apr 2026 15:52:04 +0200 Subject: [PATCH 5/5] fix(k3d): expose gateway on NodePort 31994/31993 for external workers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cubic review finding (manifests/k3d/beta9.yaml around L361): The manifest's comment declared "using NodePort instead" but the only gateway Service was ClusterIP on 1994/1993/9090. Tailscale peers and external workers hitting :31994/:31993 on the OCI node saw connection refused because no NodePort Service ever materialised — matches empty output of `ss -tlnp | grep 31994` on the live host. Fix: add `Service/gateway-nodeport` (type: NodePort) alongside the existing ClusterIP Service, with: - nodePort: 31994 -> targetPort 1994 (HTTP) - nodePort: 31993 -> targetPort 1993 (gRPC) - selector: app=gateway Kept the ClusterIP Service unchanged for in-cluster traffic. Validated with `python3 -c "import yaml; yaml.safe_load_all(...)"`; kubectl dry-run not available on the build host but the Service matches the Kubernetes v1 Service schema used elsewhere in the repo. Co-Authored-By: Claudistrator --- manifests/k3d/beta9.yaml | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/manifests/k3d/beta9.yaml b/manifests/k3d/beta9.yaml index ecb6cdf87..f3f81edec 100644 --- a/manifests/k3d/beta9.yaml +++ b/manifests/k3d/beta9.yaml @@ -194,6 +194,36 @@ spec: targetPort: 9090 protocol: TCP --- +# External NodePort Service so Tailscale peers (SDK, workers, tests) can reach +# the gateway via the host's stable :31994 (HTTP) / :31993 (gRPC) ports. The +# internal ClusterIP Service `gateway` above stays for in-cluster traffic. +# Cubic review finding: the previous manifest referenced 31994/31993 in docs +# but never materialised a Service, so `ss -tlnp | grep 31994` returned empty +# on the OCI host. +apiVersion: v1 +kind: Service +metadata: + name: gateway-nodeport + namespace: beta9 + labels: + app: gateway + tier: external +spec: + type: NodePort + selector: + app: gateway + ports: + - name: http + port: 1994 + targetPort: 1994 + nodePort: 31994 + protocol: TCP + - name: grpc + port: 1993 + targetPort: 1993 + nodePort: 31993 + protocol: TCP +--- # NOTE: IngressRouteTCP removed - using NodePort instead (no Traefik dependency) --- apiVersion: helm.cattle.io/v1