fix(sessionwire): publish negotiated ceilings through a guard

The preamble is lazy: whichever goroutine first uses a connection establishes it
and publishes the negotiated ceilings under a once. Capabilities does not go
through that once, so it read the same field without any ordering against the
write. A real offline composition caught it under the race detector, writing
from the receive goroutine while the attachment worker built its Hello.

The client and the server both guarded nothing here; both now publish the
ceilings under a mutex and read them under the same mutex. Send and receive
paths keep reading after ensurePreamble returns, which orders the publication
before their read, and the limits accessor documents that.

Verified by mutation: reading the field directly again makes the new test report
the data race, and the app-level race gate that first exposed it now passes.

commit: b25a8d38a96cd1b56d5a0f3cbfeb760d653e3862
author: brice · 33d ago · 2026-09-19 12:42 UTC
parent: aea1a9b
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +127 -9

M internal/adapters/sessionwire/client.go +31 -6


@@ -16,6 +16,11 @@ import (
16	16		type clientConnection struct {
17	17		raw wire.Transport
18	18		ceilings protoCeilings
	19	+	// ceilingsMu guards ceilings. The lazy preamble publishes the negotiated
	20	+	// values from whichever goroutine first uses the connection, while another
	21	+	// goroutine may already be asking for capabilities or starting a send, so
	22	+	// the once-guarded write is not by itself a happens-before edge for them.
	23	+	ceilingsMu sync.Mutex
19	24
20	25		preambleOnce sync.Once
21	26		preambleErr error

@@ -112,8 +117,10 @@ func (c *clientConnection) ensurePreamble() error {
112	117		c.preambleOnce.Do(func() {
113	118		ctx, cancel := context.WithDeadline(context.Background(), c.deadline)
114	119		defer cancel()
115		-	c.ceilings, c.preambleErr = runProtoClientPreamble(ctx, c.raw, c.ceilings)
116		-	if c.preambleErr != nil {
	120	+	next, err := runProtoClientPreamble(ctx, c.raw, c.limits())
	121	+	c.publishLimits(next)
	122	+	c.preambleErr = err
	123	+	if err != nil {
117	124		_ = c.raw.Close()
118	125		}
119	126		c.finishPreamble()

@@ -122,6 +129,21 @@ func (c *clientConnection) ensurePreamble() error {
122	129		return c.preambleErr
123	130		}
124	131
	132	+	// limits returns the negotiated ceilings under the lock the preamble publishes
	133	+	// them through.
	134	+	func (c *clientConnection) limits() protoCeilings {
	135	+	c.ceilingsMu.Lock()
	136	+	defer c.ceilingsMu.Unlock()
	137	+	return c.ceilings
	138	+	}
	139	+
	140	+	// publishLimits records the ceilings one preamble negotiation produced.
	141	+	func (c *clientConnection) publishLimits(next protoCeilings) {
	142	+	c.ceilingsMu.Lock()
	143	+	c.ceilings = next
	144	+	c.ceilingsMu.Unlock()
	145	+	}
	146	+
125	147		func (c *clientConnection) SendClient(message protocol.ClientMessage) error {
126	148		if err := c.ensurePreamble(); err != nil {
127	149		return err

@@ -134,7 +156,7 @@ func (c *clientConnection) SendClient(message protocol.ClientMessage) error {
134	156		if err != nil {
135	157		return err
136	158		}
137		-	if err := checkCategoryCeiling(raw, clientEnvelopeCategory(raw), c.ceilings.maxReceiveEnvelopeBytes); err != nil {
	159	+	if err := checkCategoryCeiling(raw, clientEnvelopeCategory(raw), c.limits().maxReceiveEnvelopeBytes); err != nil {
138	160		return err
139	161		}
140	162		return c.raw.Send(wire.Envelope{Payload: raw})

@@ -148,7 +170,8 @@ func (c *clientConnection) ReceiveServer() (protocol.ServerMessage, error) {
148	170		if err != nil {
149	171		return nil, err
150	172		}
151		-	message, decodeErr := decodeServerEnvelopeWithin(envelope.Payload, c.ceilings.maxReceiveEnvelopeBytes, c.ceilings.outputDataLimit)
	173	+	limits := c.limits()
	174	+	message, decodeErr := decodeServerEnvelopeWithin(envelope.Payload, limits.maxReceiveEnvelopeBytes, limits.outputDataLimit)
152	175		if decodeErr == nil {
153	176		return message, nil
154	177		}

@@ -156,7 +179,7 @@ func (c *clientConnection) ReceiveServer() (protocol.ServerMessage, error) {
156	179		}
157	180
158	181		func (c *clientConnection) Capabilities() protocol.ConnectionCapabilities {
159		-	return rawCapabilities(c.raw, c.ceilings.outputDataLimit)
	182	+	return rawCapabilities(c.raw, c.limits().outputDataLimit)
160	183		}
161	184
162	185		func (c *clientConnection) LinkState() ports.LinkState { return rawLinkState(c.raw) }

@@ -192,7 +215,9 @@ func (c *clientConnection) ensurePreambleWith(ctx context.Context) error {
192	215		c.preambleOnce.Do(func() {
193	216		ctx, cancel := context.WithDeadline(ctx, c.deadline)
194	217		defer cancel()
195		-	c.ceilings, c.preambleErr = runProtoClientPreamble(ctx, c.raw, c.ceilings)
	218	+	next, err := runProtoClientPreamble(ctx, c.raw, c.limits())
	219	+	c.publishLimits(next)
	220	+	c.preambleErr = err
196	221		c.finishPreamble()
197	222		})
198	223		c.runHandshakeHooks()

 type clientConnection struct {
 	raw      wire.Transport
 	ceilings protoCeilings
+	// ceilingsMu guards ceilings. The lazy preamble publishes the negotiated
+	// values from whichever goroutine first uses the connection, while another
+	// goroutine may already be asking for capabilities or starting a send, so
+	// the once-guarded write is not by itself a happens-before edge for them.
+	ceilingsMu sync.Mutex
 	preambleOnce sync.Once
 	preambleErr  error
 	c.preambleOnce.Do(func() {
 		ctx, cancel := context.WithDeadline(context.Background(), c.deadline)
 		defer cancel()
-		c.ceilings, c.preambleErr = runProtoClientPreamble(ctx, c.raw, c.ceilings)
+		next, err := runProtoClientPreamble(ctx, c.raw, c.limits())
-		if c.preambleErr != nil {
+		c.publishLimits(next)
+		c.preambleErr = err
+		if err != nil {
 			_ = c.raw.Close()
 		}
 		c.finishPreamble()
 	return c.preambleErr
 }
+// limits returns the negotiated ceilings under the lock the preamble publishes
+// them through.
+func (c *clientConnection) limits() protoCeilings {
+	c.ceilingsMu.Lock()
+	defer c.ceilingsMu.Unlock()
+	return c.ceilings
+}
+// publishLimits records the ceilings one preamble negotiation produced.
+func (c *clientConnection) publishLimits(next protoCeilings) {
+	c.ceilingsMu.Lock()
+	c.ceilings = next
+	c.ceilingsMu.Unlock()
+}
 func (c *clientConnection) SendClient(message protocol.ClientMessage) error {
 	if err := c.ensurePreamble(); err != nil {
 		return err
 	if err != nil {
 		return err
 	}
-	if err := checkCategoryCeiling(raw, clientEnvelopeCategory(raw), c.ceilings.maxReceiveEnvelopeBytes); err != nil {
+	if err := checkCategoryCeiling(raw, clientEnvelopeCategory(raw), c.limits().maxReceiveEnvelopeBytes); err != nil {
 		return err
 	}
 	return c.raw.Send(wire.Envelope{Payload: raw})
 	if err != nil {
 		return nil, err
 	}
-	message, decodeErr := decodeServerEnvelopeWithin(envelope.Payload, c.ceilings.maxReceiveEnvelopeBytes, c.ceilings.outputDataLimit)
+	limits := c.limits()
+	message, decodeErr := decodeServerEnvelopeWithin(envelope.Payload, limits.maxReceiveEnvelopeBytes, limits.outputDataLimit)
 	if decodeErr == nil {
 		return message, nil
 	}
 }
 func (c *clientConnection) Capabilities() protocol.ConnectionCapabilities {
-	return rawCapabilities(c.raw, c.ceilings.outputDataLimit)
+	return rawCapabilities(c.raw, c.limits().outputDataLimit)
 }
 func (c *clientConnection) LinkState() ports.LinkState         { return rawLinkState(c.raw) }
 	c.preambleOnce.Do(func() {
 		ctx, cancel := context.WithDeadline(ctx, c.deadline)
 		defer cancel()
-		c.ceilings, c.preambleErr = runProtoClientPreamble(ctx, c.raw, c.ceilings)
+		next, err := runProtoClientPreamble(ctx, c.raw, c.limits())
+		c.publishLimits(next)
+		c.preambleErr = err
 		c.finishPreamble()
 	})
 	c.runHandshakeHooks()

A internal/adapters/sessionwire/preamble_race_test.go +70 -0

@@ -0,0 +1,70 @@
	1	+	package sessionwire
	2	+
	3	+	import (
	4	+	"io"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+
	9	+	"github.com/bnema/vev/internal/protocol/wire"
	10	+	)
	11	+
	12	+	// failingTransport fails every receive, so the lazy preamble settles quickly
	13	+	// while still writing the ceilings it negotiated.
	14	+	type failingTransport struct{}
	15	+
	16	+	func (failingTransport) Send(wire.Envelope) error { return nil }
	17	+	func (failingTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
	18	+	func (failingTransport) Close() error { return nil }
	19	+
	20	+	// TestCapabilitiesReadIsSafeDuringPreamble proves a caller may ask for
	21	+	// capabilities while another goroutine is establishing the preamble.
	22	+	//
	23	+	// The preamble is lazy: whoever first uses the connection runs it, and it
	24	+	// publishes the negotiated ceilings. A capabilities read does not go through
	25	+	// that once, so it needs the same lock the publication takes; without it the
	26	+	// read races with the write, which is exactly what a real offline composition
	27	+	// reported under the race detector.
	28	+	func TestCapabilitiesReadIsSafeDuringPreamble(t *testing.T) {
	29	+	connection := NewClientConnection(failingTransport{})
	30	+	require.NotNil(t, connection)
	31	+
	32	+	done := make(chan struct{})
	33	+	go func() {
	34	+	defer close(done)
	35	+	_, _ = connection.ReceiveServer()
	36	+	}()
	37	+
	38	+	// Read capabilities for the whole window in which the preamble publishes.
	39	+	for {
	40	+	select {
	41	+	case <-done:
	42	+	return
	43	+	default:
	44	+	_ = connection.Capabilities()
	45	+	}
	46	+	}
	47	+	}
	48	+
	49	+	// TestServerCapabilitiesReadIsSafeDuringPreamble is the server-side twin: the
	50	+	// daemon-facing connection has the same lazy publication and the same unlocked
	51	+	// capabilities read.
	52	+	func TestServerCapabilitiesReadIsSafeDuringPreamble(t *testing.T) {
	53	+	connection, ok := NewServerConnection(failingTransport{}).(*serverConnection)
	54	+	require.True(t, ok)
	55	+
	56	+	done := make(chan struct{})
	57	+	go func() {
	58	+	defer close(done)
	59	+	_, _ = connection.ReceiveClient()
	60	+	}()
	61	+
	62	+	for {
	63	+	select {
	64	+	case <-done:
	65	+	return
	66	+	default:
	67	+	_ = connection.Capabilities()
	68	+	}
	69	+	}
	70	+	}

@@ -0,0 +1,70 @@
	1	+	package sessionwire
	2	+
	3	+	import (
	4	+	"io"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+
	9	+	"github.com/bnema/vev/internal/protocol/wire"
	10	+	)
	11	+
	12	+	// failingTransport fails every receive, so the lazy preamble settles quickly
	13	+	// while still writing the ceilings it negotiated.
	14	+	type failingTransport struct{}
	15	+
	16	+	func (failingTransport) Send(wire.Envelope) error { return nil }
	17	+	func (failingTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
	18	+	func (failingTransport) Close() error { return nil }
	19	+
	20	+	// TestCapabilitiesReadIsSafeDuringPreamble proves a caller may ask for
	21	+	// capabilities while another goroutine is establishing the preamble.
	22	+	//
	23	+	// The preamble is lazy: whoever first uses the connection runs it, and it
	24	+	// publishes the negotiated ceilings. A capabilities read does not go through
	25	+	// that once, so it needs the same lock the publication takes; without it the
	26	+	// read races with the write, which is exactly what a real offline composition
	27	+	// reported under the race detector.
	28	+	func TestCapabilitiesReadIsSafeDuringPreamble(t *testing.T) {
	29	+	connection := NewClientConnection(failingTransport{})
	30	+	require.NotNil(t, connection)
	31	+
	32	+	done := make(chan struct{})
	33	+	go func() {
	34	+	defer close(done)
	35	+	_, _ = connection.ReceiveServer()
	36	+	}()
	37	+
	38	+	// Read capabilities for the whole window in which the preamble publishes.
	39	+	for {
	40	+	select {
	41	+	case <-done:
	42	+	return
	43	+	default:
	44	+	_ = connection.Capabilities()
	45	+	}
	46	+	}
	47	+	}
	48	+
	49	+	// TestServerCapabilitiesReadIsSafeDuringPreamble is the server-side twin: the
	50	+	// daemon-facing connection has the same lazy publication and the same unlocked
	51	+	// capabilities read.
	52	+	func TestServerCapabilitiesReadIsSafeDuringPreamble(t *testing.T) {
	53	+	connection, ok := NewServerConnection(failingTransport{}).(*serverConnection)
	54	+	require.True(t, ok)
	55	+
	56	+	done := make(chan struct{})
	57	+	go func() {
	58	+	defer close(done)
	59	+	_, _ = connection.ReceiveClient()
	60	+	}()
	61	+
	62	+	for {
	63	+	select {
	64	+	case <-done:
	65	+	return
	66	+	default:
	67	+	_ = connection.Capabilities()
	68	+	}
	69	+	}
	70	+	}

M internal/adapters/sessionwire/server.go +26 -3


@@ -24,6 +24,11 @@ var (
24	24		type serverConnection struct {
25	25		raw wire.Transport
26	26		ceilings protoCeilings
	27	+	// ceilingsMu guards ceilings. The lazy preamble publishes the negotiated
	28	+	// values from whichever goroutine first uses the connection, while another
	29	+	// goroutine may already be asking for capabilities, so the once-guarded write
	30	+	// is not by itself a happens-before edge for that reader.
	31	+	ceilingsMu sync.Mutex
27	32
28	33		preambleOnce sync.Once
29	34		preambleErr error

@@ -137,8 +142,10 @@ func (c *serverConnection) ensurePreamble() error {
137	142		c.preambleOnce.Do(func() {
138	143		ctx, cancel := context.WithDeadline(context.Background(), c.deadline)
139	144		defer cancel()
140		-	c.ceilings, c.preambleErr = runProtoServerPreamble(ctx, c.raw, c.ceilings)
141		-	if c.preambleErr != nil {
	145	+	next, err := runProtoServerPreamble(ctx, c.raw, c.limits())
	146	+	c.publishLimits(next)
	147	+	c.preambleErr = err
	148	+	if err != nil {
142	149		_ = c.raw.Close()
143	150		}
144	151		c.finishPreamble()

@@ -147,6 +154,22 @@ func (c *serverConnection) ensurePreamble() error {
147	154		return c.preambleErr
148	155		}
149	156
	157	+	// limits returns the negotiated ceilings under the lock the preamble publishes
	158	+	// them through. Send paths may read them without the lock once ensurePreamble
	159	+	// has returned, because that call orders the publish before the read.
	160	+	func (c *serverConnection) limits() protoCeilings {
	161	+	c.ceilingsMu.Lock()
	162	+	defer c.ceilingsMu.Unlock()
	163	+	return c.ceilings
	164	+	}
	165	+
	166	+	// publishLimits records the ceilings one preamble negotiation produced.
	167	+	func (c *serverConnection) publishLimits(next protoCeilings) {
	168	+	c.ceilingsMu.Lock()
	169	+	c.ceilings = next
	170	+	c.ceilingsMu.Unlock()
	171	+	}
	172	+
150	173		func (c *serverConnection) ReceiveClient() (protocol.ClientMessage, error) {
151	174		if err := c.ensurePreamble(); err != nil {
152	175		return nil, c.preambleDecodeFailure(err)

@@ -439,7 +462,7 @@ func clientFailureFor(envelope wire.ClientEnvelope, err error) protocol.Decode
439	462		}
440	463
441	464		func (c *serverConnection) Capabilities() protocol.ConnectionCapabilities {
442		-	return rawCapabilities(c.raw, c.ceilings.outputDataLimit)
	465	+	return rawCapabilities(c.raw, c.limits().outputDataLimit)
443	466		}
444	467
445	468		func (c *serverConnection) LinkState() ports.LinkState { return rawLinkState(c.raw) }

 type serverConnection struct {
 	raw      wire.Transport
 	ceilings protoCeilings
+	// ceilingsMu guards ceilings. The lazy preamble publishes the negotiated
+	// values from whichever goroutine first uses the connection, while another
+	// goroutine may already be asking for capabilities, so the once-guarded write
+	// is not by itself a happens-before edge for that reader.
+	ceilingsMu sync.Mutex
 	preambleOnce sync.Once
 	preambleErr  error
 	c.preambleOnce.Do(func() {
 		ctx, cancel := context.WithDeadline(context.Background(), c.deadline)
 		defer cancel()
-		c.ceilings, c.preambleErr = runProtoServerPreamble(ctx, c.raw, c.ceilings)
+		next, err := runProtoServerPreamble(ctx, c.raw, c.limits())
-		if c.preambleErr != nil {
+		c.publishLimits(next)
+		c.preambleErr = err
+		if err != nil {
 			_ = c.raw.Close()
 		}
 		c.finishPreamble()
 	return c.preambleErr
 }
+// limits returns the negotiated ceilings under the lock the preamble publishes
+// them through. Send paths may read them without the lock once ensurePreamble
+// has returned, because that call orders the publish before the read.
+func (c *serverConnection) limits() protoCeilings {
+	c.ceilingsMu.Lock()
+	defer c.ceilingsMu.Unlock()
+	return c.ceilings
+}
+// publishLimits records the ceilings one preamble negotiation produced.
+func (c *serverConnection) publishLimits(next protoCeilings) {
+	c.ceilingsMu.Lock()
+	c.ceilings = next
+	c.ceilingsMu.Unlock()
+}
 func (c *serverConnection) ReceiveClient() (protocol.ClientMessage, error) {
 	if err := c.ensurePreamble(); err != nil {
 		return nil, c.preambleDecodeFailure(err)
 }
 func (c *serverConnection) Capabilities() protocol.ConnectionCapabilities {
-	return rawCapabilities(c.raw, c.ceilings.outputDataLimit)
+	return rawCapabilities(c.raw, c.limits().outputDataLimit)
 }
 func (c *serverConnection) LinkState() ports.LinkState         { return rawLinkState(c.raw) }