Ilker ManapandClaude Opus 5
ec1b6b04df
CI / test (push) Successful in 54s
Two bugs found while verifying mTLS against real machines. session cancelled its own derived context and then asked that context whether the agent was shutting down. The answer was always yes, so every stream error read as a clean close: the log said err=<nil>, the backoff reset, and an agent whose certificate the control plane refused hammered it forever at the base interval with nothing to diagnose. It now keeps the parent context to answer that question, and a refusal reaches the operator naming both the certificate and the claim. Cancelling the context also did not stop the agent. HTTP/2 watches the context while the request body is being written, and a bidirectional stream spends its life blocked on the read side instead, so the read never returned. Measured on a deployed node: `systemctl stop llmgw-agent` took 90 seconds and ended in SIGKILL, on every restart. Closing the response when the context ends is what actually unblocks it. TestAgentStopsWhenContextEnds hangs reliably without the second fix, which is the only reason to trust that it tests it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
60 lines
1.7 KiB
Go
60 lines
1.7 KiB
Go
package agent
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
|
|
)
|
|
|
|
// TestAgentStopsWhenContextEnds pins down something cancellation does not do by
|
|
// itself. A bidirectional stream spends its life blocked reading, and HTTP/2
|
|
// stops watching the context once the request headers are away, so an agent
|
|
// that only cancels keeps reading until something kills it. Under systemd that
|
|
// is the full stop timeout on every restart, followed by SIGKILL.
|
|
func TestAgentStopsWhenContextEnds(t *testing.T) {
|
|
serverTLS, clientTLS := pki(t, "llmgw-node01")
|
|
svc, inv, url := startTLS(t, serverTLS)
|
|
|
|
cfg := DefaultConfig()
|
|
cfg.NodeID = "llmgw-node01"
|
|
cfg.Server = url
|
|
cfg.TLS = clientTLS
|
|
cfg.HeartbeatInterval = 50 * time.Millisecond
|
|
cfg.Discovery.SkipGPUs = true
|
|
cfg.Discovery.IBRoot = filepath.Join(t.TempDir(), "absent")
|
|
cfg.Superv = supervisor.Config{
|
|
WorkDir: t.TempDir(), LogLines: 8,
|
|
StopTimeout: time.Second, HealthInterval: 20 * time.Millisecond,
|
|
}
|
|
|
|
client, err := Client(cfg.TLS, 5*time.Second)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ag, err := New(cfg, client, slog.New(slog.DiscardHandler))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { _ = ag.Run(ctx); close(done) }()
|
|
|
|
// Cancelling before the stream is established would prove nothing: the
|
|
// blocked read is the case that matters.
|
|
if !eventually(t, 10*time.Second, registered(svc, inv, "llmgw-node01")) {
|
|
t.Fatal("the agent never registered")
|
|
}
|
|
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the agent did not stop within 5s of being told to")
|
|
}
|
|
}
|