From 0343c7445f0baa176f094d70aaa1a455c1e8ece1 Mon Sep 17 00:00:00 2001 From: Werner Stein Date: Fri, 16 Oct 2026 11:52:17 +0200 Subject: [PATCH 1/3] refactor(hosts): factor the session deadline into a pure function The retry clamp in session could only be exercised through a live socket. A pure sessionDeadline lets tests pin each branch directly. No behaviour change. Refs: #84 Co-Authored-By: Claude Sonnet 5.5 --- apps/host_agent/internal/connection/client.go | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/apps/host_agent/internal/connection/client.go b/apps/host_agent/internal/connection/client.go index a22c82b4..9fbe0425 100644 --- a/apps/host_agent/internal/connection/client.go +++ b/apps/host_agent/internal/connection/client.go @@ -265,19 +265,25 @@ var ( renewRetry = 30 * time.Second ) -func session(ctx context.Context, state *identity.State, limit time.Time, onEvent func(Event)) error { - expires, e := identity.Expires(state) - if e != nil { - return ErrTransport - } - // End an idle socket at actual expiry, or at renewal time, independently of IO. +// sessionDeadline ends an idle socket at renewal time (six hours before expiry), +// or at actual expiry when renewal is already due, and never later than limit. +func sessionDeadline(now, expires, limit time.Time) time.Time { deadline := expires.Add(-6 * time.Hour) - if !deadline.After(time.Now()) { + if !deadline.After(now) { deadline = expires } if !limit.IsZero() && limit.Before(deadline) { deadline = limit } + return deadline +} + +func session(ctx context.Context, state *identity.State, limit time.Time, onEvent func(Event)) error { + expires, e := identity.Expires(state) + if e != nil { + return ErrTransport + } + deadline := sessionDeadline(time.Now(), expires, limit) ctx, cancel := context.WithDeadline(ctx, deadline) defer cancel() c, e := client(state, false) From 9e8bcf21d23d580257ec0c18fd2432c8e49f6718 Mon Sep 17 00:00:00 2001 From: Werner Stein Date: Fri, 16 Oct 2026 11:52:17 +0200 Subject: [PATCH 2/3] test(hosts): cover the renewal loop and the validity deadline Run's renewal branches and the clamp that bounds a session after a failed renewal had no direct tests; deleting the clamp still passed. Drive Run against a loopback TLS/WebSocket server for renewal success, renewal failure on a valid identity and reload failure, and end a blocked real ws.Read at the deadline. Rename the watchValidity test to match what it proves and register the new IDs. Refs: #84 Co-Authored-By: Claude Sonnet 5.5 --- .../internal/connection/client_test.go | 2 +- .../internal/connection/loopback_test.go | 224 ++++++++++++++++++ docs/development/ci-acceptance.json | 8 +- 3 files changed, 232 insertions(+), 2 deletions(-) create mode 100644 apps/host_agent/internal/connection/loopback_test.go diff --git a/apps/host_agent/internal/connection/client_test.go b/apps/host_agent/internal/connection/client_test.go index 9041af3f..91d4d03a 100644 --- a/apps/host_agent/internal/connection/client_test.go +++ b/apps/host_agent/internal/connection/client_test.go @@ -151,7 +151,7 @@ func savedPendingDir(t *testing.T) (string, *identity.State, identity.Response) return dir, state, response } -func TestExpiryWatchdogCancelsBlockedIO(t *testing.T) { +func TestWatchValidityCancelsOnInvalidIdentity(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() var valid atomic.Bool diff --git a/apps/host_agent/internal/connection/loopback_test.go b/apps/host_agent/internal/connection/loopback_test.go new file mode 100644 index 00000000..e5a2a155 --- /dev/null +++ b/apps/host_agent/internal/connection/loopback_test.go @@ -0,0 +1,224 @@ +package connection + +import ( + "context" + "encoding/json" + "encoding/pem" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/identity" +) + +func TestSessionDeadlineClampsToRetryLimit(t *testing.T) { + now := time.Now() + expires := now.Add(24 * time.Hour) + if got := sessionDeadline(now, expires, time.Time{}); !got.Equal(expires.Add(-6 * time.Hour)) { + t.Fatalf("no limit: deadline = %v, want renewal time", got) + } + if got := sessionDeadline(now, now.Add(time.Hour), time.Time{}); !got.Equal(now.Add(time.Hour)) { + t.Fatalf("renewal due: deadline = %v, want expiry", got) + } + limit := now.Add(30 * time.Second) + if got := sessionDeadline(now, expires, limit); !got.Equal(limit) { + t.Fatalf("limit before renewal time: deadline = %v, want limit", got) + } + if got := sessionDeadline(now, now.Add(10*time.Second), limit); !got.Equal(now.Add(10 * time.Second)) { + t.Fatalf("limit after expiry must not extend the session: %v", got) + } +} + +// loopbackAgent serves the agent API over TLS on loopback. A nil activate fails activation. +func loopbackAgent(t *testing.T, activate http.HandlerFunc, ws func(*websocket.Conn)) (*httptest.Server, *atomic.Int32) { + t.Helper() + var activations atomic.Int32 + mux := http.NewServeMux() + mux.HandleFunc("/agent/v1/identity/activate", func(w http.ResponseWriter, r *http.Request) { + activations.Add(1) + if activate == nil { + http.Error(w, "no", http.StatusInternalServerError) + return + } + activate(w, r) + }) + mux.HandleFunc("/agent/v1/connect", func(w http.ResponseWriter, r *http.Request) { + c, e := websocket.Accept(w, r, &websocket.AcceptOptions{CompressionMode: websocket.CompressionDisabled}) + if e != nil { + return + } + defer c.CloseNow() + ws(c) + }) + srv := httptest.NewTLSServer(mux) + t.Cleanup(srv.Close) + return srv, &activations +} + +// handshake acknowledges hello and the first heartbeat, then stays silent until the peer closes. +func handshake(closed chan<- struct{}) func(*websocket.Conn) { + var once sync.Once + return func(c *websocket.Conn) { + ctx := context.Background() + for _, ack := range []string{"hello_ack", "heartbeat_ack"} { + _, raw, e := c.Read(ctx) + if e != nil { + return + } + var m map[string]any + if json.Unmarshal(raw, &m) != nil { + return + } + reply, _ := json.Marshal(map[string]any{"schema_version": 1, "kind": ack, "owner_id": m["owner_id"], "host_id": m["host_id"]}) + if c.Write(ctx, websocket.MessageText, reply) != nil { + return + } + } + _, _, _ = c.Read(ctx) + if closed != nil { + once.Do(func() { close(closed) }) + } + } +} + +// loopbackDir returns an identity dir with an active credential and a saved issued pending one, +// pointed at the loopback server. +func loopbackDir(t *testing.T, srv *httptest.Server) (string, *identity.State, identity.Response) { + t.Helper() + dir, state, response := savedPendingDir(t) + ca := filepath.Join(filepath.Dir(dir), "server-ca.pem") + block := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srv.Certificate().Raw}) + if e := os.WriteFile(ca, block, 0600); e != nil { + t.Fatal(e) + } + state.Endpoint, state.ServerCAFile = srv.URL, ca + if e := identity.Save(dir, state); e != nil { + t.Fatal(e) + } + return dir, state, response +} + +func activateWith(response *identity.Response) http.HandlerFunc { + return func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(response) + } +} + +// runUntilConnected drives Run until its first connected event, then stops it. +func runUntilConnected(t *testing.T, dir string) error { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + connected := make(chan struct{}, 1) + done := make(chan error, 1) + go func() { + done <- Run(ctx, dir, func(e Event) { + if e.Kind == "connected" { + select { + case connected <- struct{}{}: + default: + } + } + }) + }() + select { + case <-connected: + cancel() + return <-done + case e := <-done: + t.Fatalf("Run returned before connecting: %v", e) + case <-time.After(15 * time.Second): + t.Fatal("Run did not connect") + } + return nil +} + +func TestRunRenewsOverLoopbackThenConnects(t *testing.T) { + var response identity.Response + srv, activations := loopbackAgent(t, activateWith(&response), handshake(nil)) + dir, _, r := loopbackDir(t, srv) + response = r + if e := runUntilConnected(t, dir); e != nil { + t.Fatal(e) + } + if activations.Load() != 1 { + t.Fatalf("activations = %d", activations.Load()) + } + state, e := identity.Load(dir) + if e != nil || state.Active.IdentityID != r.IdentityID { + t.Fatalf("renewed identity not promoted: %v", e) + } +} + +func TestRunStaysConnectedWhenRenewalFailsOnValidIdentity(t *testing.T) { + srv, activations := loopbackAgent(t, nil, handshake(nil)) + dir, before, r := loopbackDir(t, srv) + if e := runUntilConnected(t, dir); e != nil { + t.Fatal(e) + } + if activations.Load() == 0 { + t.Fatal("renewal was not attempted") + } + state, e := identity.Load(dir) + if e != nil || state.Active.IdentityID != before.Active.IdentityID || state.Active.IdentityID == r.IdentityID { + t.Fatalf("failed renewal changed the active identity: %v", e) + } +} + +func TestRunReturnsTransportWhenReloadFailsAfterRenewal(t *testing.T) { + oldRenew := renewFn + defer func() { renewFn = oldRenew }() + dir := "" + renewFn = func(context.Context, string) error { + return os.WriteFile(filepath.Join(dir, "identity.json"), []byte("{"), 0600) + } + srv, _ := loopbackAgent(t, nil, handshake(nil)) + dir, _, _ = loopbackDir(t, srv) + done := make(chan error, 1) + go func() { done <- Run(context.Background(), dir, nil) }() + select { + case e := <-done: + if e != ErrTransport { + t.Fatalf("Run = %v, want ErrTransport", e) + } + case <-time.After(10 * time.Second): + t.Fatal("Run did not return after failed reload") + } +} + +func TestSessionEndsBlockedReadAtValidityDeadline(t *testing.T) { + closed := make(chan struct{}) + srv, _ := loopbackAgent(t, nil, handshake(closed)) + dir, _, _ := loopbackDir(t, srv) + state, e := identity.Load(dir) + if e != nil { + t.Fatal(e) + } + // The credential is valid for hours; only the limit can end the idle socket. + start := time.Now() + done := make(chan error, 1) + go func() { done <- session(context.Background(), state, start.Add(1500*time.Millisecond), nil) }() + select { + case e = <-done: + if e != ErrTransport { + t.Fatalf("session = %v, want ErrTransport", e) + } + case <-time.After(8 * time.Second): + t.Fatal("session did not end at the deadline") + } + if time.Since(start) < time.Second { + t.Fatalf("session ended early: %v", time.Since(start)) + } + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Fatal("socket was not closed at the deadline") + } +} diff --git a/docs/development/ci-acceptance.json b/docs/development/ci-acceptance.json index 7ea0d607..81724ccf 100644 --- a/docs/development/ci-acceptance.json +++ b/docs/development/ci-acceptance.json @@ -568,8 +568,14 @@ "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestEnrollmentCodeAndUnsafeStateFailBeforeTransport", "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestActivationRejectsIncompleteOrMismatchedIssuedIdentity", "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestSharedHostConnectionFixtures", - "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestExpiryWatchdogCancelsBlockedIO", + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestWatchValidityCancelsOnInvalidIdentity", "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestAcknowledgementSequenceFailsClosed", + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestRunRetriesFailedRenewalWhileConnected", + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestSessionDeadlineClampsToRetryLimit", + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestRunRenewsOverLoopbackThenConnects", + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestRunStaysConnectedWhenRenewalFailsOnValidIdentity", + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestRunReturnsTransportWhenReloadFailsAfterRenewal", + "github.com/wstein/workharbor-nxt/apps/host_agent/internal/connection/TestSessionEndsBlockedReadAtValidityDeadline", "github.com/wstein/workharbor-nxt/apps/host_agent/internal/identity/TestAtomicIdentityPersistenceFailuresAndRestartTransitions", "github.com/wstein/workharbor-nxt/apps/host_agent/internal/identity/TestInitialPossessionDurabilityFailureNeverReportsSuccess", "github.com/wstein/workharbor-nxt/apps/host_agent/cmd/workharbor-agent/TestPrivateDockerProxyModeRejectsMalformedLifetimeAndSocket", From 6189c0afee9f740231eebf3ebc26f15461e1187b Mon Sep 17 00:00:00 2001 From: Werner Stein Date: Fri, 16 Oct 2026 13:02:07 +0200 Subject: [PATCH 3/3] test(hosts): document two loopback test limits The reload-failure test keeps its server only for the identity fixture, and the blocked-read test proves the socket close, not a direct CloseNow call, since context cancel also closes it. Refs: #84 Co-Authored-By: Claude Sonnet 5.5 --- apps/host_agent/internal/connection/loopback_test.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/apps/host_agent/internal/connection/loopback_test.go b/apps/host_agent/internal/connection/loopback_test.go index e5a2a155..90f9bdb7 100644 --- a/apps/host_agent/internal/connection/loopback_test.go +++ b/apps/host_agent/internal/connection/loopback_test.go @@ -179,6 +179,8 @@ func TestRunReturnsTransportWhenReloadFailsAfterRenewal(t *testing.T) { renewFn = func(context.Context, string) error { return os.WriteFile(filepath.Join(dir, "identity.json"), []byte("{"), 0600) } + // The server is never contacted: renewFn is stubbed. It stays only because + // loopbackDir needs a server certificate to build a valid identity fixture. srv, _ := loopbackAgent(t, nil, handshake(nil)) dir, _, _ = loopbackDir(t, srv) done := make(chan error, 1) @@ -216,6 +218,8 @@ func TestSessionEndsBlockedReadAtValidityDeadline(t *testing.T) { if time.Since(start) < time.Second { t.Fatalf("session ended early: %v", time.Since(start)) } + // This proves the socket is closed, not that session called CloseNow: coder/websocket + // also closes the connection when its context is cancelled. select { case <-closed: case <-time.After(5 * time.Second):