a73x

05ed8adf

test(syncsvc): a deadline nothing outlives proves nothing

a73x   2026-08-23 11:11

Commit message
test(syncsvc): a deadline nothing outlives proves nothing

The console handshake deadline was pinned by a 50ms sleep against a 10s
timeout, so deleting either the set or the clear left the test green.
handshakeTimeout becomes a Service field — consoleHandshakeTimeout stays
its production value — so a test can inject a bound it can actually
outlive: the bridging test now round-trips after the deadline would have
fired, and a new test holds a silent agent open and requires OpenTCP to
give up rather than park the caller's goroutine for good.

TestApplyReportSkipsUnchangedLeases named the skip but only read the
cache the skip consults; the skip itself is counted by
TestNetTrackerWritesOnceThenStaysQuiet. Renamed to the claim it can
make on its own — that applyReport routes every reported lease through
that cache, keyed by VM, rather than writing straight to the store.
internal/server/syncsvc/network_test.go
Old New
@@ -141,24 +141,45 @@ func TestApplyReportKeepsTheLastKnownLease(t *testing.T) {
141 assert.Equal(t, "192.168.0.99", vm.NetworkIP, "a good address after junk still lands") 141 assert.Equal(t, "192.168.0.99", vm.NetworkIP, "a good address after junk still lands")
142 } 142 }
143 143
144 // TestApplyReportSkipsUnchangedLeases proves the per-VM dedup: a bridged guest 144 // TestApplyReportRoutesLeasesThroughTheDedupCache pins the half of the skip
145 // re-reports the same address every tick forever, and the store runs on one 145 // that lives in applyReport: every reported address goes through netIPTrack,
146 // connection. Only a change may cost a write. 146 // keyed by VM. A bridged guest re-reports the same lease every tick forever and
147 func TestApplyReportSkipsUnchangedLeases(t *testing.T) { 147 // the store runs on one connection, so a direct RecordVMNetworkIP here would
148 // spend a round trip per bridged guest per tick for the life of the fleet — and
149 // leave no trace, because the write succeeds every time.
150 //
151 // The skip itself — repeat writes nothing, change writes — belongs to
152 // TestNetTrackerWritesOnceThenStaysQuiet, which counts writes. This test can
153 // only see the cache, so it asserts what the cache proves: that applyReport
154 // consults it, with the reported address under the reporting VM's id.
155 func TestApplyReportRoutesLeasesThroughTheDedupCache(t *testing.T) {
148 f := setup(t) 156 f := setup(t)
149 networkedVM(t, f, "vm1") 157 networkedVM(t, f, "vm1")
150 rep := &pb.Report{Vms: []*pb.VMStatus{{ 158 report := func(ip string) *pb.Report {
151 VmId: "vm1", PowerState: "running", Phase: "ready", Ip: "10.77.1.2", NetworkIp: "192.168.0.42", 159 return &pb.Report{Vms: []*pb.VMStatus{{
152 }}} 160 VmId: "vm1", PowerState: "running", Phase: "ready", Ip: "10.77.1.2", NetworkIp: ip,
161 }}}
162 }
153 163
154 f.svc.applyReport(f.host.ID, rep) 164 f.svc.applyReport(f.host.ID, report("192.168.0.42"))
155 f.svc.applyReport(f.host.ID, rep) 165 f.svc.applyReport(f.host.ID, report("192.168.0.42"))
156 f.svc.applyReport(f.host.ID, rep) 166 f.svc.applyReport(f.host.ID, report("192.168.0.42"))
167 assertLeaseCache(t, f, map[string]string{"vm1": "192.168.0.42"},
168 "a lease that never reached the cache is a lease re-written every tick forever")
169
170 // A renewal to a different address must land in the cache, or the guest's
171 // new lease is remembered as the old one and never written again.
172 f.svc.applyReport(f.host.ID, report("192.168.0.99"))
173 assertLeaseCache(t, f, map[string]string{"vm1": "192.168.0.99"},
174 "the cache must hold the newest reported lease, not the first one seen")
175 }
157 176
177 // assertLeaseCache reads netIPTrack under its own lock and compares it to want.
178 func assertLeaseCache(t *testing.T, f *fixture, want map[string]string, msg string) {
179 t.Helper()
158 f.svc.netIPTrack.mu.Lock() 180 f.svc.netIPTrack.mu.Lock()
159 defer f.svc.netIPTrack.mu.Unlock() 181 defer f.svc.netIPTrack.mu.Unlock()
160 assert.Equal(t, map[string]string{"vm1": "192.168.0.42"}, f.svc.netIPTrack.last, 182 assert.Equal(t, want, f.svc.netIPTrack.last, msg)
161 "the address is remembered once, so the repeats write nothing")
162 } 183 }
163 184
164 // TestApplyReportForgetsAReapedVMsLease bounds the cache: a VM whose row is 185 // TestApplyReportForgetsAReapedVMsLease bounds the cache: a VM whose row is
internal/server/syncsvc/syncsvc.go
Old New
@@ -64,6 +64,11 @@ type Service struct {
64 maxCredAge time.Duration 64 maxCredAge time.Duration
65 // writeTimeout bounds each down-stream snapshot write (see defaultWriteTimeout). 65 // writeTimeout bounds each down-stream snapshot write (see defaultWriteTimeout).
66 writeTimeout time.Duration 66 writeTimeout time.Duration
67 // handshakeTimeout bounds the open/opened exchange on a server-initiated
68 // stream (see consoleHandshakeTimeout, its production value). A field, not
69 // the constant, so a test can inject a timeout it can actually outlive:
70 // nothing can prove the deadline is set — or cleared — against ten seconds.
71 handshakeTimeout time.Duration
67 // consoleMu guards conns: the live QUIC connection per agent, registered 72 // consoleMu guards conns: the live QUIC connection per agent, registered
68 // after auth in handleConn and deregistered when the session ends. The 73 // after auth in handleConn and deregistered when the session ends. The
69 // console broker opens per-session streams on it. 74 // console broker opens per-session streams on it.
@@ -103,7 +108,8 @@ func newWithWriteTimeout(st *store.Store, reg *registry.Registry, h *hub.Hub, se
103 writeTimeout = defaultWriteTimeout 108 writeTimeout = defaultWriteTimeout
104 } 109 }
105 return &Service{st: st, reg: reg, hub: h, secret: secret, maxCredAge: maxCredAge, writeTimeout: writeTimeout, 110 return &Service{st: st, reg: reg, hub: h, secret: secret, maxCredAge: maxCredAge, writeTimeout: writeTimeout,
106 conns: map[string]quic.Connection{}, recorder: st, tracker: newStatusTracker(), netTrack: newNetTracker(), 111 handshakeTimeout: consoleHandshakeTimeout,
112 conns: map[string]quic.Connection{}, recorder: st, tracker: newStatusTracker(), netTrack: newNetTracker(),
107 uplinkTrack: newNetTracker(), certTrack: newNetTracker(), netIPTrack: newNetTracker(), 113 uplinkTrack: newNetTracker(), certTrack: newNetTracker(), netIPTrack: newNetTracker(),
108 offers: map[string]offer{}, now: time.Now} 114 offers: map[string]offer{}, now: time.Now}
109 } 115 }
@@ -664,9 +670,10 @@ func toRegistryMetrics(m *pb.HostMetrics) registry.Metrics {
664 // ErrAgentOffline reports that the target host has no live sync connection. 670 // ErrAgentOffline reports that the target host has no live sync connection.
665 var ErrAgentOffline = errors.New("agent not connected") 671 var ErrAgentOffline = errors.New("agent not connected")
666 672
667 // consoleHandshakeTimeout bounds the ConsoleOpen/ConsoleOpened exchange so a 673 // consoleHandshakeTimeout is the production value of Service.handshakeTimeout:
668 // wedged agent cannot pin the WS handler. The bridged session itself has no 674 // it bounds the ConsoleOpen/ConsoleOpened exchange so a wedged agent cannot pin
669 // deadline — consoles are long-lived. 675 // the WS handler. The bridged session itself has no deadline — consoles are
676 // long-lived.
670 const consoleHandshakeTimeout = 10 * time.Second 677 const consoleHandshakeTimeout = 10 * time.Second
671 678
672 // openStream is the shared server-initiated stream procedure behind OpenConsole 679 // openStream is the shared server-initiated stream procedure behind OpenConsole
@@ -678,8 +685,8 @@ const consoleHandshakeTimeout = 10 * time.Second
678 // returned Close tears down both directions. 685 // returned Close tears down both directions.
679 // 686 //
680 // ctx bounds stream OPENING only; the handshake that follows is bounded by 687 // ctx bounds stream OPENING only; the handshake that follows is bounded by
681 // consoleHandshakeTimeout instead, so a call can outlive ctx cancellation by 688 // s.handshakeTimeout instead, so a call can outlive ctx cancellation by up to
682 // up to that long (10s) before returning. 689 // that long (10s in production) before returning.
683 func (s *Service) openStream(ctx context.Context, hostID, label string, open *pb.ServerMessage, 690 func (s *Service) openStream(ctx context.Context, hostID, label string, open *pb.ServerMessage,
684 checkReply func(*pb.AgentMessage) (ok bool, reason string, present bool)) (io.ReadWriteCloser, error) { 691 checkReply func(*pb.AgentMessage) (ok bool, reason string, present bool)) (io.ReadWriteCloser, error) {
685 s.consoleMu.Lock() 692 s.consoleMu.Lock()
@@ -693,7 +700,7 @@ func (s *Service) openStream(ctx context.Context, hostID, label string, open *pb
693 return nil, fmt.Errorf("open %s stream: %w", label, err) 700 return nil, fmt.Errorf("open %s stream: %w", label, err)
694 } 701 }
695 cs := consoleStream{st} 702 cs := consoleStream{st}
696 if err := st.SetDeadline(time.Now().Add(consoleHandshakeTimeout)); err != nil { 703 if err := st.SetDeadline(time.Now().Add(s.handshakeTimeout)); err != nil {
697 cs.Close() 704 cs.Close()
698 return nil, err 705 return nil, err
699 } 706 }
internal/server/syncsvc/syncsvc_test.go
Old New
@@ -699,12 +699,16 @@ func TestOpenTCPNoAgent(t *testing.T) {
699 } 699 }
700 700
701 // TestOpenTCPBridgesBytes drives the full TCPOpen/TCPOpened handshake against a 701 // TestOpenTCPBridgesBytes drives the full TCPOpen/TCPOpened handshake against a
702 // fake agent that answers ok=true, then asserts a usable RWC is returned, bytes 702 // fake agent that answers ok=true, then asserts a usable RWC is returned and
703 // pass both directions, and — after the handshake — the stream carries no 703 // bytes pass both directions.
704 // residual read deadline (a delayed read still succeeds, proving the session is 704 //
705 // long-lived, mirroring OpenConsole's SetDeadline(time.Time{}) clear). 705 // The round trip happens after a pause LONGER than the injected handshake
706 // timeout, which is what makes it evidence that the deadline was cleared: a
707 // residual deadline would kill every console and every tunnel a fixed time
708 // after it opened, and sessions here are long-lived by definition.
706 func TestOpenTCPBridgesBytes(t *testing.T) { 709 func TestOpenTCPBridgesBytes(t *testing.T) {
707 f := setup(t) 710 f := setup(t)
711 f.svc.handshakeTimeout = 200 * time.Millisecond
708 c := mustDial(t, f) 712 c := mustDial(t, f)
709 c.recv(t) // initial snapshot 713 c.recv(t) // initial snapshot
710 714
@@ -744,17 +748,56 @@ func TestOpenTCPBridgesBytes(t *testing.T) {
744 748
745 assert.Equal(t, uint32(8080), <-gotPort, "agent must receive the requested port") 749 assert.Equal(t, uint32(8080), <-gotPort, "agent must receive the requested port")
746 750
747 // The handshake deadline (10s) must have been cleared: a delayed write/read 751 // Outlive the handshake deadline, then round-trip.
748 // well short of that still round-trips. 752 time.Sleep(300 * time.Millisecond)
749 time.Sleep(50 * time.Millisecond)
750 _, err = rwc.Write([]byte("ping")) 753 _, err = rwc.Write([]byte("ping"))
751 require.NoError(t, err) 754 require.NoError(t, err, "the handshake deadline must be cleared: a session that dies once it elapses is a console that drops mid-use")
752 buf := make([]byte, 4) 755 buf := make([]byte, 4)
753 _, err = io.ReadFull(rwc, buf) 756 _, err = io.ReadFull(rwc, buf)
754 require.NoError(t, err) 757 require.NoError(t, err, "the handshake deadline must be cleared: a session that dies once it elapses is a console that drops mid-use")
755 assert.Equal(t, "ping", string(buf)) 758 assert.Equal(t, "ping", string(buf))
756 } 759 }
757 760
761 // TestOpenTCPAbandonsAWedgedAgent pins the other half of the deadline's life:
762 // an agent that accepts the stream and never answers must not hold the caller.
763 // Without the deadline the WS handler goroutine waits forever on a read no one
764 // will satisfy — a silent leak, one goroutine per attempted console.
765 func TestOpenTCPAbandonsAWedgedAgent(t *testing.T) {
766 f := setup(t)
767 f.svc.handshakeTimeout = 100 * time.Millisecond
768 c := mustDial(t, f)
769 c.recv(t) // initial snapshot
770
771 accepted := make(chan struct{})
772 go func() {
773 st, err := c.conn.AcceptStream(context.Background())
774 if err != nil {
775 return
776 }
777 var open pb.ServerMessage
778 if transport.ReadMsg(st, &open, transport.DefaultMaxFrame) != nil {
779 return
780 }
781 close(accepted) // the open arrived; the reply never will
782 <-time.After(10 * time.Second)
783 }()
784
785 done := make(chan error, 1)
786 go func() {
787 _, err := f.svc.OpenTCP(context.Background(), f.host.ID, "vm-1", 22)
788 done <- err
789 }()
790
791 <-accepted
792 select {
793 case err := <-done:
794 require.Error(t, err, "a silent agent must fail the open, not return a stream nobody is on the far end of")
795 assert.Contains(t, err.Error(), "tcp reply", "the failure must name the read that timed out")
796 case <-time.After(3 * time.Second):
797 t.Fatal("OpenTCP is still waiting on a wedged agent: without the handshake deadline this goroutine never returns, and one leaks per attempted session")
798 }
799 }
800
758 // TestOpenTCPRefused pins the ok=false path: the agent replies with an error 801 // TestOpenTCPRefused pins the ok=false path: the agent replies with an error
759 // string and OpenTCP surfaces it (not a usable stream). 802 // string and OpenTCP surfaces it (not a usable stream).
760 func TestOpenTCPRefused(t *testing.T) { 803 func TestOpenTCPRefused(t *testing.T) {