refactor(sessionwire): share the preamble lifecycle between client and server
A single embedded handshake owns the once-only preamble, the fixed deadline, the negotiated ceilings, the completion signal and its hooks. Client and server connections only choose their directional preamble runner.
- commit
- 48eff4cf49a380d76377d154107bc4738abc4c35
- author
- brice · · 2026-10-08 05:08 UTC
- parent
- 8e56f16
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +171 -259
M internal/adapters/sessionwire/client.go +3 -116
| @@ -1,9 +1,7 @@ | |||
| 1 | 1 | package sessionwire | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | - | "context" | |
| 5 | 4 | "errors" | |
| 6 | - | "sync" | |
| 7 | 5 | "time" | |
| 8 | 6 | ||
| 9 | 7 | "google.golang.org/protobuf/proto" | |
| @@ -14,27 +12,7 @@ import ( | |||
| 14 | 12 | ) | |
| 15 | 13 | ||
| 16 | 14 | type clientConnection struct { | |
| 17 | - | raw wire.Transport | |
| 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 | |
| 24 | - | ||
| 25 | - | preambleOnce sync.Once | |
| 26 | - | preambleErr error | |
| 27 | - | deadline time.Time | |
| 28 | - | ||
| 29 | - | // preambleDone closes exactly once when the handshake has run, whether it | |
| 30 | - | // succeeded or failed. It backs the handshake completion hook that a logical | |
| 31 | - | // mux connection exposes; the absolute deadline is never restarted. | |
| 32 | - | preambleDone chan struct{} | |
| 33 | - | ||
| 34 | - | hooksMu sync.Mutex | |
| 35 | - | handshakeDone bool | |
| 36 | - | hooksRun bool | |
| 37 | - | hooks []func() | |
| 15 | + | handshake | |
| 38 | 16 | } | |
| 39 | 17 | ||
| 40 | 18 | var _ ports.ClientConnection = (*clientConnection)(nil) | |
| @@ -46,102 +24,11 @@ func NewClientConnection(raw wire.Transport) ports.ClientConnection { | |||
| 46 | 24 | if raw == nil { | |
| 47 | 25 | return nil | |
| 48 | 26 | } | |
| 49 | - | return &clientConnection{raw: raw, ceilings: defaultProtoCeilings(), deadline: time.Now().Add(protocol.HandshakeTimeout), preambleDone: make(chan struct{})} | |
| 50 | - | } | |
| 51 | - | ||
| 52 | - | // HandshakeDeadline returns the absolute local deadline of the session | |
| 53 | - | // handshake this connection started with. It is the one accepted deadline the | |
| 54 | - | // preamble, Hello/Welcome, and first committed publication share; the value is | |
| 55 | - | // fixed for the connection's lifetime and is never restarted. | |
| 56 | - | func (c *clientConnection) HandshakeDeadline() time.Time { return c.deadline } | |
| 57 | - | ||
| 58 | - | // HandshakeDone returns a channel that is closed exactly once when the | |
| 59 | - | // handshake has run, whether it succeeded or failed. | |
| 60 | - | func (c *clientConnection) HandshakeDone() <-chan struct{} { return c.preambleDone } | |
| 61 | - | ||
| 62 | - | // OnHandshakeComplete registers fn to run exactly once when the handshake has | |
| 63 | - | // run. Registering after completion runs fn immediately; a nil fn is ignored. | |
| 64 | - | func (c *clientConnection) OnHandshakeComplete(fn func()) { | |
| 65 | - | if fn == nil { | |
| 66 | - | return | |
| 67 | - | } | |
| 68 | - | c.hooksMu.Lock() | |
| 69 | - | if c.handshakeDone { | |
| 70 | - | c.hooksMu.Unlock() | |
| 71 | - | fn() | |
| 72 | - | return | |
| 73 | - | } | |
| 74 | - | c.hooks = append(c.hooks, fn) | |
| 75 | - | c.hooksMu.Unlock() | |
| 76 | - | } | |
| 77 | - | ||
| 78 | - | // finishPreamble publishes handshake completion from inside the preamble's | |
| 79 | - | // sync.Once: it closes the done channel and marks the handshake complete so a | |
| 80 | - | // later registration runs immediately. It deliberately runs no registered hook, | |
| 81 | - | // because a hook that calls back into the connection would re-enter sync.Once | |
| 82 | - | // and deadlock; runHandshakeHooks runs the hooks after the Once body returns. | |
| 83 | - | // It tolerates a zero-value connection that never allocated the done channel | |
| 84 | - | // (test literals and seam-only construction), so closing is nil-safe. | |
| 85 | - | func (c *clientConnection) finishPreamble() { | |
| 86 | - | if c.preambleDone != nil { | |
| 87 | - | close(c.preambleDone) | |
| 88 | - | } | |
| 89 | - | c.hooksMu.Lock() | |
| 90 | - | c.handshakeDone = true | |
| 91 | - | c.hooksMu.Unlock() | |
| 92 | - | } | |
| 93 | - | ||
| 94 | - | // runHandshakeHooks runs the completion hooks registered before the handshake | |
| 95 | - | // finished exactly once, after the preamble's sync.Once body has returned. | |
| 96 | - | // Running them outside the Once is what lets a hook call back into the | |
| 97 | - | // connection - for example to send a message - without re-entering sync.Once | |
| 98 | - | // and deadlocking. Every ensurePreamble caller invokes it; only the first | |
| 99 | - | // drains the hooks, and a registrant that arrives after completion observes the | |
| 100 | - | // completed flag and runs itself. | |
| 101 | - | func (c *clientConnection) runHandshakeHooks() { | |
| 102 | - | c.hooksMu.Lock() | |
| 103 | - | if c.hooksRun { | |
| 104 | - | c.hooksMu.Unlock() | |
| 105 | - | return | |
| 106 | - | } | |
| 107 | - | c.hooksRun = true | |
| 108 | - | hooks := c.hooks | |
| 109 | - | c.hooks = nil | |
| 110 | - | c.hooksMu.Unlock() | |
| 111 | - | for _, fn := range hooks { | |
| 112 | - | fn() | |
| 113 | - | } | |
| 27 | + | return &clientConnection{handshake: newHandshake(raw, time.Now().Add(protocol.HandshakeTimeout))} | |
| 114 | 28 | } | |
| 115 | 29 | ||
| 116 | 30 | func (c *clientConnection) ensurePreamble() error { | |
| 117 | - | c.preambleOnce.Do(func() { | |
| 118 | - | ctx, cancel := context.WithDeadline(context.Background(), c.deadline) | |
| 119 | - | defer cancel() | |
| 120 | - | next, err := runProtoClientPreamble(ctx, c.raw, c.limits()) | |
| 121 | - | c.publishLimits(next) | |
| 122 | - | c.preambleErr = err | |
| 123 | - | if err != nil { | |
| 124 | - | _ = c.raw.Close() | |
| 125 | - | } | |
| 126 | - | c.finishPreamble() | |
| 127 | - | }) | |
| 128 | - | c.runHandshakeHooks() | |
| 129 | - | return c.preambleErr | |
| 130 | - | } | |
| 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() | |
| 31 | + | return c.handshake.ensurePreamble(runProtoClientPreamble) | |
| 145 | 32 | } | |
| 146 | 33 | ||
| 147 | 34 | func (c *clientConnection) SendClient(message protocol.ClientMessage) error { | |
| @@ -1,9 +1,7 @@ | |||||
| 1 | package sessionwire | 1 | package sessionwire | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | - | "context" | |||
| 5 | "errors" | 4 | "errors" | ||
| 6 | - | "sync" | |||
| 7 | "time" | 5 | "time" | ||
| 8 | 6 | ||||
| 9 | "google.golang.org/protobuf/proto" | 7 | "google.golang.org/protobuf/proto" | ||
| @@ -14,27 +12,7 @@ import ( | |||||
| 14 | ) | 12 | ) | ||
| 15 | 13 | ||||
| 16 | type clientConnection struct { | 14 | type clientConnection struct { | ||
| 17 | - | raw wire.Transport | 15 | + | handshake |
| 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 | |||
| 24 | - | ||||
| 25 | - | preambleOnce sync.Once | |||
| 26 | - | preambleErr error | |||
| 27 | - | deadline time.Time | |||
| 28 | - | ||||
| 29 | - | // preambleDone closes exactly once when the handshake has run, whether it | |||
| 30 | - | // succeeded or failed. It backs the handshake completion hook that a logical | |||
| 31 | - | // mux connection exposes; the absolute deadline is never restarted. | |||
| 32 | - | preambleDone chan struct{} | |||
| 33 | - | ||||
| 34 | - | hooksMu sync.Mutex | |||
| 35 | - | handshakeDone bool | |||
| 36 | - | hooksRun bool | |||
| 37 | - | hooks []func() | |||
| 38 | } | 16 | } | ||
| 39 | 17 | ||||
| 40 | var _ ports.ClientConnection = (*clientConnection)(nil) | 18 | var _ ports.ClientConnection = (*clientConnection)(nil) | ||
| @@ -46,102 +24,11 @@ func NewClientConnection(raw wire.Transport) ports.ClientConnection { | |||||
| 46 | if raw == nil { | 24 | if raw == nil { | ||
| 47 | return nil | 25 | return nil | ||
| 48 | } | 26 | } | ||
| 49 | - | return &clientConnection{raw: raw, ceilings: defaultProtoCeilings(), deadline: time.Now().Add(protocol.HandshakeTimeout), preambleDone: make(chan struct{})} | 27 | + | return &clientConnection{handshake: newHandshake(raw, time.Now().Add(protocol.HandshakeTimeout))} |
| 50 | - | } | |||
| 51 | - | ||||
| 52 | - | // HandshakeDeadline returns the absolute local deadline of the session | |||
| 53 | - | // handshake this connection started with. It is the one accepted deadline the | |||
| 54 | - | // preamble, Hello/Welcome, and first committed publication share; the value is | |||
| 55 | - | // fixed for the connection's lifetime and is never restarted. | |||
| 56 | - | func (c *clientConnection) HandshakeDeadline() time.Time { return c.deadline } | |||
| 57 | - | ||||
| 58 | - | // HandshakeDone returns a channel that is closed exactly once when the | |||
| 59 | - | // handshake has run, whether it succeeded or failed. | |||
| 60 | - | func (c *clientConnection) HandshakeDone() <-chan struct{} { return c.preambleDone } | |||
| 61 | - | ||||
| 62 | - | // OnHandshakeComplete registers fn to run exactly once when the handshake has | |||
| 63 | - | // run. Registering after completion runs fn immediately; a nil fn is ignored. | |||
| 64 | - | func (c *clientConnection) OnHandshakeComplete(fn func()) { | |||
| 65 | - | if fn == nil { | |||
| 66 | - | return | |||
| 67 | - | } | |||
| 68 | - | c.hooksMu.Lock() | |||
| 69 | - | if c.handshakeDone { | |||
| 70 | - | c.hooksMu.Unlock() | |||
| 71 | - | fn() | |||
| 72 | - | return | |||
| 73 | - | } | |||
| 74 | - | c.hooks = append(c.hooks, fn) | |||
| 75 | - | c.hooksMu.Unlock() | |||
| 76 | - | } | |||
| 77 | - | ||||
| 78 | - | // finishPreamble publishes handshake completion from inside the preamble's | |||
| 79 | - | // sync.Once: it closes the done channel and marks the handshake complete so a | |||
| 80 | - | // later registration runs immediately. It deliberately runs no registered hook, | |||
| 81 | - | // because a hook that calls back into the connection would re-enter sync.Once | |||
| 82 | - | // and deadlock; runHandshakeHooks runs the hooks after the Once body returns. | |||
| 83 | - | // It tolerates a zero-value connection that never allocated the done channel | |||
| 84 | - | // (test literals and seam-only construction), so closing is nil-safe. | |||
| 85 | - | func (c *clientConnection) finishPreamble() { | |||
| 86 | - | if c.preambleDone != nil { | |||
| 87 | - | close(c.preambleDone) | |||
| 88 | - | } | |||
| 89 | - | c.hooksMu.Lock() | |||
| 90 | - | c.handshakeDone = true | |||
| 91 | - | c.hooksMu.Unlock() | |||
| 92 | - | } | |||
| 93 | - | ||||
| 94 | - | // runHandshakeHooks runs the completion hooks registered before the handshake | |||
| 95 | - | // finished exactly once, after the preamble's sync.Once body has returned. | |||
| 96 | - | // Running them outside the Once is what lets a hook call back into the | |||
| 97 | - | // connection - for example to send a message - without re-entering sync.Once | |||
| 98 | - | // and deadlocking. Every ensurePreamble caller invokes it; only the first | |||
| 99 | - | // drains the hooks, and a registrant that arrives after completion observes the | |||
| 100 | - | // completed flag and runs itself. | |||
| 101 | - | func (c *clientConnection) runHandshakeHooks() { | |||
| 102 | - | c.hooksMu.Lock() | |||
| 103 | - | if c.hooksRun { | |||
| 104 | - | c.hooksMu.Unlock() | |||
| 105 | - | return | |||
| 106 | - | } | |||
| 107 | - | c.hooksRun = true | |||
| 108 | - | hooks := c.hooks | |||
| 109 | - | c.hooks = nil | |||
| 110 | - | c.hooksMu.Unlock() | |||
| 111 | - | for _, fn := range hooks { | |||
| 112 | - | fn() | |||
| 113 | - | } | |||
| 114 | } | 28 | } | ||
| 115 | 29 | ||||
| 116 | func (c *clientConnection) ensurePreamble() error { | 30 | func (c *clientConnection) ensurePreamble() error { | ||
| 117 | - | c.preambleOnce.Do(func() { | 31 | + | return c.handshake.ensurePreamble(runProtoClientPreamble) |
| 118 | - | ctx, cancel := context.WithDeadline(context.Background(), c.deadline) | |||
| 119 | - | defer cancel() | |||
| 120 | - | next, err := runProtoClientPreamble(ctx, c.raw, c.limits()) | |||
| 121 | - | c.publishLimits(next) | |||
| 122 | - | c.preambleErr = err | |||
| 123 | - | if err != nil { | |||
| 124 | - | _ = c.raw.Close() | |||
| 125 | - | } | |||
| 126 | - | c.finishPreamble() | |||
| 127 | - | }) | |||
| 128 | - | c.runHandshakeHooks() | |||
| 129 | - | return c.preambleErr | |||
| 130 | - | } | |||
| 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 | } | 32 | } | ||
| 146 | 33 | ||||
| 147 | func (c *clientConnection) SendClient(message protocol.ClientMessage) error { | 34 | func (c *clientConnection) SendClient(message protocol.ClientMessage) error { | ||
M internal/adapters/sessionwire/command_result_test.go +2 -2
| @@ -227,11 +227,11 @@ func TestCommandResultIsBoundedControlEnvelope(t *testing.T) { | |||
| 227 | 227 | require.ErrorIs(t, checkCategoryCeiling(oversizeRaw, categoryControl, wire.AbsoluteEnvelopeLimit), wire.ErrScanLength, | |
| 228 | 228 | "a CommandResult above the control ceiling must fail even under an absolute negotiated ceiling") | |
| 229 | 229 | ||
| 230 | - | send := &serverConnection{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()} | |
| 230 | + | send := &serverConnection{handshake: handshake{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()}} | |
| 231 | 231 | require.NoError(t, send.SendServer(message)) | |
| 232 | 232 | require.ErrorIs(t, send.SendServer(oversize), wire.ErrScanLength) | |
| 233 | 233 | ||
| 234 | - | recv := &clientConnection{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversizeRaw}}}, ceilings: defaultProtoCeilings()} | |
| 234 | + | recv := &clientConnection{handshake: handshake{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversizeRaw}}}, ceilings: defaultProtoCeilings()}} | |
| 235 | 235 | recv.preambleOnce.Do(func() {}) | |
| 236 | 236 | _, err = recv.ReceiveServer() | |
| 237 | 237 | require.Error(t, err) | |
| @@ -227,11 +227,11 @@ func TestCommandResultIsBoundedControlEnvelope(t *testing.T) { | |||||
| 227 | require.ErrorIs(t, checkCategoryCeiling(oversizeRaw, categoryControl, wire.AbsoluteEnvelopeLimit), wire.ErrScanLength, | 227 | require.ErrorIs(t, checkCategoryCeiling(oversizeRaw, categoryControl, wire.AbsoluteEnvelopeLimit), wire.ErrScanLength, | ||
| 228 | "a CommandResult above the control ceiling must fail even under an absolute negotiated ceiling") | 228 | "a CommandResult above the control ceiling must fail even under an absolute negotiated ceiling") | ||
| 229 | 229 | ||||
| 230 | - | send := &serverConnection{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()} | 230 | + | send := &serverConnection{handshake: handshake{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()}} |
| 231 | require.NoError(t, send.SendServer(message)) | 231 | require.NoError(t, send.SendServer(message)) | ||
| 232 | require.ErrorIs(t, send.SendServer(oversize), wire.ErrScanLength) | 232 | require.ErrorIs(t, send.SendServer(oversize), wire.ErrScanLength) | ||
| 233 | 233 | ||||
| 234 | - | recv := &clientConnection{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversizeRaw}}}, ceilings: defaultProtoCeilings()} | 234 | + | recv := &clientConnection{handshake: handshake{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversizeRaw}}}, ceilings: defaultProtoCeilings()}} |
| 235 | recv.preambleOnce.Do(func() {}) | 235 | recv.preambleOnce.Do(func() {}) | ||
| 236 | _, err = recv.ReceiveServer() | 236 | _, err = recv.ReceiveServer() | ||
| 237 | require.Error(t, err) | 237 | require.Error(t, err) | ||
A internal/adapters/sessionwire/handshake.go +142 -0
| @@ -0,0 +1,142 @@ | |||
| 1 | + | package sessionwire | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "sync" | |
| 6 | + | "time" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/protocol/wire" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | // preambleRunner runs one directional preamble and returns the ceilings it | |
| 12 | + | // negotiated. | |
| 13 | + | type preambleRunner func(context.Context, wire.Transport, protoCeilings) (protoCeilings, error) | |
| 14 | + | ||
| 15 | + | // handshake is the lazy, once-only preamble lifecycle shared by client and | |
| 16 | + | // server connections: the fixed absolute deadline, the negotiated ceilings, | |
| 17 | + | // the completion signal, and the completion hooks. | |
| 18 | + | type handshake struct { | |
| 19 | + | raw wire.Transport | |
| 20 | + | ceilings protoCeilings | |
| 21 | + | // ceilingsMu guards ceilings. The lazy preamble publishes the negotiated | |
| 22 | + | // values from whichever goroutine first uses the connection, while another | |
| 23 | + | // goroutine may already be asking for capabilities or starting a send, so | |
| 24 | + | // the once-guarded write is not by itself a happens-before edge for them. | |
| 25 | + | ceilingsMu sync.Mutex | |
| 26 | + | ||
| 27 | + | preambleOnce sync.Once | |
| 28 | + | preambleErr error | |
| 29 | + | deadline time.Time | |
| 30 | + | ||
| 31 | + | // preambleDone closes exactly once when the handshake has run, whether it | |
| 32 | + | // succeeded or failed. It backs the handshake completion seam a logical mux | |
| 33 | + | // connection exposes; the absolute deadline is never restarted. | |
| 34 | + | preambleDone chan struct{} | |
| 35 | + | ||
| 36 | + | hooksMu sync.Mutex | |
| 37 | + | handshakeDone bool | |
| 38 | + | hooksRun bool | |
| 39 | + | hooks []func() | |
| 40 | + | } | |
| 41 | + | ||
| 42 | + | func newHandshake(raw wire.Transport, deadline time.Time) handshake { | |
| 43 | + | return handshake{raw: raw, ceilings: defaultProtoCeilings(), deadline: deadline, preambleDone: make(chan struct{})} | |
| 44 | + | } | |
| 45 | + | ||
| 46 | + | // HandshakeDeadline returns the absolute local deadline of the session | |
| 47 | + | // handshake this connection started with. It is the one accepted deadline the | |
| 48 | + | // preamble, Hello/Welcome, and first committed publication share; the value is | |
| 49 | + | // fixed for the connection's lifetime and is never restarted. | |
| 50 | + | func (h *handshake) HandshakeDeadline() time.Time { return h.deadline } | |
| 51 | + | ||
| 52 | + | // HandshakeDone returns a channel that is closed exactly once when the | |
| 53 | + | // handshake has run, whether it succeeded or failed. | |
| 54 | + | func (h *handshake) HandshakeDone() <-chan struct{} { return h.preambleDone } | |
| 55 | + | ||
| 56 | + | // OnHandshakeComplete registers fn to run exactly once when the handshake has | |
| 57 | + | // run. Registering after completion runs fn immediately; a nil fn is ignored. | |
| 58 | + | func (h *handshake) OnHandshakeComplete(fn func()) { | |
| 59 | + | if fn == nil { | |
| 60 | + | return | |
| 61 | + | } | |
| 62 | + | h.hooksMu.Lock() | |
| 63 | + | if h.handshakeDone { | |
| 64 | + | h.hooksMu.Unlock() | |
| 65 | + | fn() | |
| 66 | + | return | |
| 67 | + | } | |
| 68 | + | h.hooks = append(h.hooks, fn) | |
| 69 | + | h.hooksMu.Unlock() | |
| 70 | + | } | |
| 71 | + | ||
| 72 | + | // ensurePreamble runs the preamble once, bounded by the handshake deadline. | |
| 73 | + | // A failed preamble closes the raw transport. | |
| 74 | + | func (h *handshake) ensurePreamble(run preambleRunner) error { | |
| 75 | + | h.preambleOnce.Do(func() { | |
| 76 | + | ctx, cancel := context.WithDeadline(context.Background(), h.deadline) | |
| 77 | + | defer cancel() | |
| 78 | + | next, err := run(ctx, h.raw, h.limits()) | |
| 79 | + | h.publishLimits(next) | |
| 80 | + | h.preambleErr = err | |
| 81 | + | if err != nil { | |
| 82 | + | _ = h.raw.Close() | |
| 83 | + | } | |
| 84 | + | h.finishPreamble() | |
| 85 | + | }) | |
| 86 | + | h.runHandshakeHooks() | |
| 87 | + | return h.preambleErr | |
| 88 | + | } | |
| 89 | + | ||
| 90 | + | // finishPreamble publishes handshake completion from inside the preamble's | |
| 91 | + | // sync.Once: it closes the done channel and marks the handshake complete so a | |
| 92 | + | // later registration runs immediately. It deliberately runs no registered hook, | |
| 93 | + | // because a hook that calls back into the connection would re-enter sync.Once | |
| 94 | + | // and deadlock; runHandshakeHooks runs the hooks after the Once body returns. | |
| 95 | + | // It tolerates a zero-value handshake that never allocated the done channel | |
| 96 | + | // (test literals and seam-only construction), so closing is nil-safe. | |
| 97 | + | func (h *handshake) finishPreamble() { | |
| 98 | + | if h.preambleDone != nil { | |
| 99 | + | close(h.preambleDone) | |
| 100 | + | } | |
| 101 | + | h.hooksMu.Lock() | |
| 102 | + | h.handshakeDone = true | |
| 103 | + | h.hooksMu.Unlock() | |
| 104 | + | } | |
| 105 | + | ||
| 106 | + | // runHandshakeHooks runs the completion hooks registered before the handshake | |
| 107 | + | // finished exactly once, after the preamble's sync.Once body has returned. | |
| 108 | + | // Running them outside the Once is what lets a hook call back into the | |
| 109 | + | // connection - for example to send or receive a message - without re-entering | |
| 110 | + | // sync.Once and deadlocking. Every ensurePreamble caller invokes it; only the | |
| 111 | + | // first drains the hooks, and a registrant that arrives after completion | |
| 112 | + | // observes the completed flag and runs itself. | |
| 113 | + | func (h *handshake) runHandshakeHooks() { | |
| 114 | + | h.hooksMu.Lock() | |
| 115 | + | if h.hooksRun { | |
| 116 | + | h.hooksMu.Unlock() | |
| 117 | + | return | |
| 118 | + | } | |
| 119 | + | h.hooksRun = true | |
| 120 | + | hooks := h.hooks | |
| 121 | + | h.hooks = nil | |
| 122 | + | h.hooksMu.Unlock() | |
| 123 | + | for _, fn := range hooks { | |
| 124 | + | fn() | |
| 125 | + | } | |
| 126 | + | } | |
| 127 | + | ||
| 128 | + | // limits returns the negotiated ceilings under the lock the preamble publishes | |
| 129 | + | // them through. Send paths may read them without the lock once ensurePreamble | |
| 130 | + | // has returned, because that call orders the publish before the read. | |
| 131 | + | func (h *handshake) limits() protoCeilings { | |
| 132 | + | h.ceilingsMu.Lock() | |
| 133 | + | defer h.ceilingsMu.Unlock() | |
| 134 | + | return h.ceilings | |
| 135 | + | } | |
| 136 | + | ||
| 137 | + | // publishLimits records the ceilings one preamble negotiation produced. | |
| 138 | + | func (h *handshake) publishLimits(next protoCeilings) { | |
| 139 | + | h.ceilingsMu.Lock() | |
| 140 | + | h.ceilings = next | |
| 141 | + | h.ceilingsMu.Unlock() | |
| 142 | + | } | |
| @@ -0,0 +1,142 @@ | |||||
| 1 | + | package sessionwire | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "sync" | |||
| 6 | + | "time" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/protocol/wire" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | // preambleRunner runs one directional preamble and returns the ceilings it | |||
| 12 | + | // negotiated. | |||
| 13 | + | type preambleRunner func(context.Context, wire.Transport, protoCeilings) (protoCeilings, error) | |||
| 14 | + | ||||
| 15 | + | // handshake is the lazy, once-only preamble lifecycle shared by client and | |||
| 16 | + | // server connections: the fixed absolute deadline, the negotiated ceilings, | |||
| 17 | + | // the completion signal, and the completion hooks. | |||
| 18 | + | type handshake struct { | |||
| 19 | + | raw wire.Transport | |||
| 20 | + | ceilings protoCeilings | |||
| 21 | + | // ceilingsMu guards ceilings. The lazy preamble publishes the negotiated | |||
| 22 | + | // values from whichever goroutine first uses the connection, while another | |||
| 23 | + | // goroutine may already be asking for capabilities or starting a send, so | |||
| 24 | + | // the once-guarded write is not by itself a happens-before edge for them. | |||
| 25 | + | ceilingsMu sync.Mutex | |||
| 26 | + | ||||
| 27 | + | preambleOnce sync.Once | |||
| 28 | + | preambleErr error | |||
| 29 | + | deadline time.Time | |||
| 30 | + | ||||
| 31 | + | // preambleDone closes exactly once when the handshake has run, whether it | |||
| 32 | + | // succeeded or failed. It backs the handshake completion seam a logical mux | |||
| 33 | + | // connection exposes; the absolute deadline is never restarted. | |||
| 34 | + | preambleDone chan struct{} | |||
| 35 | + | ||||
| 36 | + | hooksMu sync.Mutex | |||
| 37 | + | handshakeDone bool | |||
| 38 | + | hooksRun bool | |||
| 39 | + | hooks []func() | |||
| 40 | + | } | |||
| 41 | + | ||||
| 42 | + | func newHandshake(raw wire.Transport, deadline time.Time) handshake { | |||
| 43 | + | return handshake{raw: raw, ceilings: defaultProtoCeilings(), deadline: deadline, preambleDone: make(chan struct{})} | |||
| 44 | + | } | |||
| 45 | + | ||||
| 46 | + | // HandshakeDeadline returns the absolute local deadline of the session | |||
| 47 | + | // handshake this connection started with. It is the one accepted deadline the | |||
| 48 | + | // preamble, Hello/Welcome, and first committed publication share; the value is | |||
| 49 | + | // fixed for the connection's lifetime and is never restarted. | |||
| 50 | + | func (h *handshake) HandshakeDeadline() time.Time { return h.deadline } | |||
| 51 | + | ||||
| 52 | + | // HandshakeDone returns a channel that is closed exactly once when the | |||
| 53 | + | // handshake has run, whether it succeeded or failed. | |||
| 54 | + | func (h *handshake) HandshakeDone() <-chan struct{} { return h.preambleDone } | |||
| 55 | + | ||||
| 56 | + | // OnHandshakeComplete registers fn to run exactly once when the handshake has | |||
| 57 | + | // run. Registering after completion runs fn immediately; a nil fn is ignored. | |||
| 58 | + | func (h *handshake) OnHandshakeComplete(fn func()) { | |||
| 59 | + | if fn == nil { | |||
| 60 | + | return | |||
| 61 | + | } | |||
| 62 | + | h.hooksMu.Lock() | |||
| 63 | + | if h.handshakeDone { | |||
| 64 | + | h.hooksMu.Unlock() | |||
| 65 | + | fn() | |||
| 66 | + | return | |||
| 67 | + | } | |||
| 68 | + | h.hooks = append(h.hooks, fn) | |||
| 69 | + | h.hooksMu.Unlock() | |||
| 70 | + | } | |||
| 71 | + | ||||
| 72 | + | // ensurePreamble runs the preamble once, bounded by the handshake deadline. | |||
| 73 | + | // A failed preamble closes the raw transport. | |||
| 74 | + | func (h *handshake) ensurePreamble(run preambleRunner) error { | |||
| 75 | + | h.preambleOnce.Do(func() { | |||
| 76 | + | ctx, cancel := context.WithDeadline(context.Background(), h.deadline) | |||
| 77 | + | defer cancel() | |||
| 78 | + | next, err := run(ctx, h.raw, h.limits()) | |||
| 79 | + | h.publishLimits(next) | |||
| 80 | + | h.preambleErr = err | |||
| 81 | + | if err != nil { | |||
| 82 | + | _ = h.raw.Close() | |||
| 83 | + | } | |||
| 84 | + | h.finishPreamble() | |||
| 85 | + | }) | |||
| 86 | + | h.runHandshakeHooks() | |||
| 87 | + | return h.preambleErr | |||
| 88 | + | } | |||
| 89 | + | ||||
| 90 | + | // finishPreamble publishes handshake completion from inside the preamble's | |||
| 91 | + | // sync.Once: it closes the done channel and marks the handshake complete so a | |||
| 92 | + | // later registration runs immediately. It deliberately runs no registered hook, | |||
| 93 | + | // because a hook that calls back into the connection would re-enter sync.Once | |||
| 94 | + | // and deadlock; runHandshakeHooks runs the hooks after the Once body returns. | |||
| 95 | + | // It tolerates a zero-value handshake that never allocated the done channel | |||
| 96 | + | // (test literals and seam-only construction), so closing is nil-safe. | |||
| 97 | + | func (h *handshake) finishPreamble() { | |||
| 98 | + | if h.preambleDone != nil { | |||
| 99 | + | close(h.preambleDone) | |||
| 100 | + | } | |||
| 101 | + | h.hooksMu.Lock() | |||
| 102 | + | h.handshakeDone = true | |||
| 103 | + | h.hooksMu.Unlock() | |||
| 104 | + | } | |||
| 105 | + | ||||
| 106 | + | // runHandshakeHooks runs the completion hooks registered before the handshake | |||
| 107 | + | // finished exactly once, after the preamble's sync.Once body has returned. | |||
| 108 | + | // Running them outside the Once is what lets a hook call back into the | |||
| 109 | + | // connection - for example to send or receive a message - without re-entering | |||
| 110 | + | // sync.Once and deadlocking. Every ensurePreamble caller invokes it; only the | |||
| 111 | + | // first drains the hooks, and a registrant that arrives after completion | |||
| 112 | + | // observes the completed flag and runs itself. | |||
| 113 | + | func (h *handshake) runHandshakeHooks() { | |||
| 114 | + | h.hooksMu.Lock() | |||
| 115 | + | if h.hooksRun { | |||
| 116 | + | h.hooksMu.Unlock() | |||
| 117 | + | return | |||
| 118 | + | } | |||
| 119 | + | h.hooksRun = true | |||
| 120 | + | hooks := h.hooks | |||
| 121 | + | h.hooks = nil | |||
| 122 | + | h.hooksMu.Unlock() | |||
| 123 | + | for _, fn := range hooks { | |||
| 124 | + | fn() | |||
| 125 | + | } | |||
| 126 | + | } | |||
| 127 | + | ||||
| 128 | + | // limits returns the negotiated ceilings under the lock the preamble publishes | |||
| 129 | + | // them through. Send paths may read them without the lock once ensurePreamble | |||
| 130 | + | // has returned, because that call orders the publish before the read. | |||
| 131 | + | func (h *handshake) limits() protoCeilings { | |||
| 132 | + | h.ceilingsMu.Lock() | |||
| 133 | + | defer h.ceilingsMu.Unlock() | |||
| 134 | + | return h.ceilings | |||
| 135 | + | } | |||
| 136 | + | ||||
| 137 | + | // publishLimits records the ceilings one preamble negotiation produced. | |||
| 138 | + | func (h *handshake) publishLimits(next protoCeilings) { | |||
| 139 | + | h.ceilingsMu.Lock() | |||
| 140 | + | h.ceilings = next | |||
| 141 | + | h.ceilingsMu.Unlock() | |||
| 142 | + | } | |||
M internal/adapters/sessionwire/kill_result_test.go +2 -2
| @@ -239,11 +239,11 @@ func TestKillResultIsBoundedControlEnvelope(t *testing.T) { | |||
| 239 | 239 | require.ErrorIs(t, checkCategoryCeiling(oversize, categoryControl, wire.AbsoluteEnvelopeLimit), wire.ErrScanLength, | |
| 240 | 240 | "a KillResult above the control ceiling must fail even under an absolute negotiated ceiling") | |
| 241 | 241 | ||
| 242 | - | send := &serverConnection{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()} | |
| 242 | + | send := &serverConnection{handshake: handshake{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()}} | |
| 243 | 243 | require.NoError(t, send.SendServer(message)) | |
| 244 | 244 | require.ErrorIs(t, send.SendServer(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded, Text: string(make([]byte, wire.ControlEnvelopeLimit))}), wire.ErrScanLength) | |
| 245 | 245 | ||
| 246 | - | recv := &clientConnection{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversize}}}, ceilings: defaultProtoCeilings()} | |
| 246 | + | recv := &clientConnection{handshake: handshake{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversize}}}, ceilings: defaultProtoCeilings()}} | |
| 247 | 247 | recv.preambleOnce.Do(func() {}) | |
| 248 | 248 | _, err = recv.ReceiveServer() | |
| 249 | 249 | require.Error(t, err) | |
| @@ -239,11 +239,11 @@ func TestKillResultIsBoundedControlEnvelope(t *testing.T) { | |||||
| 239 | require.ErrorIs(t, checkCategoryCeiling(oversize, categoryControl, wire.AbsoluteEnvelopeLimit), wire.ErrScanLength, | 239 | require.ErrorIs(t, checkCategoryCeiling(oversize, categoryControl, wire.AbsoluteEnvelopeLimit), wire.ErrScanLength, | ||
| 240 | "a KillResult above the control ceiling must fail even under an absolute negotiated ceiling") | 240 | "a KillResult above the control ceiling must fail even under an absolute negotiated ceiling") | ||
| 241 | 241 | ||||
| 242 | - | send := &serverConnection{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()} | 242 | + | send := &serverConnection{handshake: handshake{raw: &scriptedTransport{}, ceilings: defaultProtoCeilings()}} |
| 243 | require.NoError(t, send.SendServer(message)) | 243 | require.NoError(t, send.SendServer(message)) | ||
| 244 | require.ErrorIs(t, send.SendServer(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded, Text: string(make([]byte, wire.ControlEnvelopeLimit))}), wire.ErrScanLength) | 244 | require.ErrorIs(t, send.SendServer(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded, Text: string(make([]byte, wire.ControlEnvelopeLimit))}), wire.ErrScanLength) | ||
| 245 | 245 | ||||
| 246 | - | recv := &clientConnection{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversize}}}, ceilings: defaultProtoCeilings()} | 246 | + | recv := &clientConnection{handshake: handshake{raw: &scriptedTransport{recv: []wire.Envelope{{Payload: oversize}}}, ceilings: defaultProtoCeilings()}} |
| 247 | recv.preambleOnce.Do(func() {}) | 247 | recv.preambleOnce.Do(func() {}) | ||
| 248 | _, err = recv.ReceiveServer() | 248 | _, err = recv.ReceiveServer() | ||
| 249 | require.Error(t, err) | 249 | require.Error(t, err) | ||
M internal/adapters/sessionwire/proto_limits_category_test.go +5 -5
| @@ -133,7 +133,7 @@ func TestClientSendCategoryEnvelopeCeilings(t *testing.T) { | |||
| 133 | 133 | for _, tt := range tests { | |
| 134 | 134 | t.Run(tt.name, func(t *testing.T) { | |
| 135 | 135 | raw := &scriptedTransport{} | |
| 136 | - | conn := &clientConnection{raw: raw, ceilings: tt.ceilings} | |
| 136 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} | |
| 137 | 137 | conn.preambleOnce.Do(func() {}) | |
| 138 | 138 | err := conn.SendClient(tt.message) | |
| 139 | 139 | if tt.wantErr != nil { | |
| @@ -194,7 +194,7 @@ func TestServerSendCategoryEnvelopeCeilings(t *testing.T) { | |||
| 194 | 194 | for _, tt := range tests { | |
| 195 | 195 | t.Run(tt.name, func(t *testing.T) { | |
| 196 | 196 | raw := &scriptedTransport{} | |
| 197 | - | conn := &serverConnection{raw: raw, ceilings: tt.ceilings} | |
| 197 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} | |
| 198 | 198 | err := conn.SendServer(tt.message) | |
| 199 | 199 | if tt.wantErr != nil { | |
| 200 | 200 | require.ErrorIs(t, err, tt.wantErr) | |
| @@ -212,7 +212,7 @@ func TestServerSendCategoryEnvelopeCeilings(t *testing.T) { | |||
| 212 | 212 | func TestServerSendModesShareControlCategoryCap(t *testing.T) { | |
| 213 | 213 | controlOver := protocol.Sessions{Sessions: []protocol.SessionInfo{{Name: strings.Repeat("x", overControlCap)}}} | |
| 214 | 214 | raw := &asyncScriptedTransport{scriptedTransport: &scriptedTransport{}} | |
| 215 | - | conn := &serverConnection{raw: raw, ceilings: defaultProtoCeilings()} | |
| 215 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: defaultProtoCeilings()}} | |
| 216 | 216 | ||
| 217 | 217 | require.ErrorIs(t, conn.SendServerAsync(controlOver), wire.ErrScanLength) | |
| 218 | 218 | require.ErrorIs(t, conn.SendServerSynchronous(controlOver), wire.ErrScanLength) | |
| @@ -331,11 +331,11 @@ func TestReceiveCategoryEnvelopeCeilings(t *testing.T) { | |||
| 331 | 331 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: tt.payload}}} | |
| 332 | 332 | var err error | |
| 333 | 333 | if tt.receiver == "server" { | |
| 334 | - | conn := &serverConnection{raw: raw, ceilings: tt.ceilings} | |
| 334 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} | |
| 335 | 335 | conn.preambleOnce.Do(func() {}) | |
| 336 | 336 | _, err = conn.ReceiveClient() | |
| 337 | 337 | } else { | |
| 338 | - | conn := &clientConnection{raw: raw, ceilings: tt.ceilings} | |
| 338 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} | |
| 339 | 339 | conn.preambleOnce.Do(func() {}) | |
| 340 | 340 | _, err = conn.ReceiveServer() | |
| 341 | 341 | } | |
| @@ -133,7 +133,7 @@ func TestClientSendCategoryEnvelopeCeilings(t *testing.T) { | |||||
| 133 | for _, tt := range tests { | 133 | for _, tt := range tests { | ||
| 134 | t.Run(tt.name, func(t *testing.T) { | 134 | t.Run(tt.name, func(t *testing.T) { | ||
| 135 | raw := &scriptedTransport{} | 135 | raw := &scriptedTransport{} | ||
| 136 | - | conn := &clientConnection{raw: raw, ceilings: tt.ceilings} | 136 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} |
| 137 | conn.preambleOnce.Do(func() {}) | 137 | conn.preambleOnce.Do(func() {}) | ||
| 138 | err := conn.SendClient(tt.message) | 138 | err := conn.SendClient(tt.message) | ||
| 139 | if tt.wantErr != nil { | 139 | if tt.wantErr != nil { | ||
| @@ -194,7 +194,7 @@ func TestServerSendCategoryEnvelopeCeilings(t *testing.T) { | |||||
| 194 | for _, tt := range tests { | 194 | for _, tt := range tests { | ||
| 195 | t.Run(tt.name, func(t *testing.T) { | 195 | t.Run(tt.name, func(t *testing.T) { | ||
| 196 | raw := &scriptedTransport{} | 196 | raw := &scriptedTransport{} | ||
| 197 | - | conn := &serverConnection{raw: raw, ceilings: tt.ceilings} | 197 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} |
| 198 | err := conn.SendServer(tt.message) | 198 | err := conn.SendServer(tt.message) | ||
| 199 | if tt.wantErr != nil { | 199 | if tt.wantErr != nil { | ||
| 200 | require.ErrorIs(t, err, tt.wantErr) | 200 | require.ErrorIs(t, err, tt.wantErr) | ||
| @@ -212,7 +212,7 @@ func TestServerSendCategoryEnvelopeCeilings(t *testing.T) { | |||||
| 212 | func TestServerSendModesShareControlCategoryCap(t *testing.T) { | 212 | func TestServerSendModesShareControlCategoryCap(t *testing.T) { | ||
| 213 | controlOver := protocol.Sessions{Sessions: []protocol.SessionInfo{{Name: strings.Repeat("x", overControlCap)}}} | 213 | controlOver := protocol.Sessions{Sessions: []protocol.SessionInfo{{Name: strings.Repeat("x", overControlCap)}}} | ||
| 214 | raw := &asyncScriptedTransport{scriptedTransport: &scriptedTransport{}} | 214 | raw := &asyncScriptedTransport{scriptedTransport: &scriptedTransport{}} | ||
| 215 | - | conn := &serverConnection{raw: raw, ceilings: defaultProtoCeilings()} | 215 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: defaultProtoCeilings()}} |
| 216 | 216 | ||||
| 217 | require.ErrorIs(t, conn.SendServerAsync(controlOver), wire.ErrScanLength) | 217 | require.ErrorIs(t, conn.SendServerAsync(controlOver), wire.ErrScanLength) | ||
| 218 | require.ErrorIs(t, conn.SendServerSynchronous(controlOver), wire.ErrScanLength) | 218 | require.ErrorIs(t, conn.SendServerSynchronous(controlOver), wire.ErrScanLength) | ||
| @@ -331,11 +331,11 @@ func TestReceiveCategoryEnvelopeCeilings(t *testing.T) { | |||||
| 331 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: tt.payload}}} | 331 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: tt.payload}}} | ||
| 332 | var err error | 332 | var err error | ||
| 333 | if tt.receiver == "server" { | 333 | if tt.receiver == "server" { | ||
| 334 | - | conn := &serverConnection{raw: raw, ceilings: tt.ceilings} | 334 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} |
| 335 | conn.preambleOnce.Do(func() {}) | 335 | conn.preambleOnce.Do(func() {}) | ||
| 336 | _, err = conn.ReceiveClient() | 336 | _, err = conn.ReceiveClient() | ||
| 337 | } else { | 337 | } else { | ||
| 338 | - | conn := &clientConnection{raw: raw, ceilings: tt.ceilings} | 338 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: tt.ceilings}} |
| 339 | conn.preambleOnce.Do(func() {}) | 339 | conn.preambleOnce.Do(func() {}) | ||
| 340 | _, err = conn.ReceiveServer() | 340 | _, err = conn.ReceiveServer() | ||
| 341 | } | 341 | } | ||
M internal/adapters/sessionwire/proto_limits_test.go +10 -10
| @@ -42,7 +42,7 @@ func TestServerSendEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 42 | 42 | ||
| 43 | 43 | t.Run("normal send", func(t *testing.T) { | |
| 44 | 44 | raw := &scriptedTransport{} | |
| 45 | - | conn := &serverConnection{raw: raw, ceilings: envelopeBound} | |
| 45 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: envelopeBound}} | |
| 46 | 46 | require.ErrorIs(t, conn.SendServer(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | |
| 47 | 47 | require.Zero(t, raw.sentLen()) | |
| 48 | 48 | conn.ceilings = outputBound | |
| @@ -51,7 +51,7 @@ func TestServerSendEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 51 | 51 | }) | |
| 52 | 52 | t.Run("async and owned synchronous send", func(t *testing.T) { | |
| 53 | 53 | raw := &asyncScriptedTransport{scriptedTransport: &scriptedTransport{}} | |
| 54 | - | conn := &serverConnection{raw: raw, ceilings: envelopeBound} | |
| 54 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: envelopeBound}} | |
| 55 | 55 | require.ErrorIs(t, conn.SendServerAsync(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | |
| 56 | 56 | require.ErrorIs(t, conn.SendServerSynchronous(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | |
| 57 | 57 | require.Empty(t, raw.async) | |
| @@ -64,7 +64,7 @@ func TestServerSendEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 64 | 64 | }) | |
| 65 | 65 | t.Run("within ceilings", func(t *testing.T) { | |
| 66 | 66 | raw := &scriptedTransport{} | |
| 67 | - | conn := &serverConnection{raw: raw, ceilings: defaultProtoCeilings()} | |
| 67 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: defaultProtoCeilings()}} | |
| 68 | 68 | require.NoError(t, conn.SendServer(protocol.Pong{})) | |
| 69 | 69 | require.NoError(t, conn.SendOutput(testOutput([]byte("12345")))) | |
| 70 | 70 | require.Equal(t, 2, raw.sentLen()) | |
| @@ -81,7 +81,7 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 81 | 81 | ||
| 82 | 82 | t.Run("server receive envelope ceiling", func(t *testing.T) { | |
| 83 | 83 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeInput}}} | |
| 84 | - | conn := &serverConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}} | |
| 84 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}}} | |
| 85 | 85 | conn.preambleOnce.Do(func() {}) | |
| 86 | 86 | _, err := conn.ReceiveClient() | |
| 87 | 87 | var failure *protocol.DecodeFailure | |
| @@ -90,7 +90,7 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 90 | 90 | }) | |
| 91 | 91 | t.Run("client receive envelope ceiling", func(t *testing.T) { | |
| 92 | 92 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeOutput}}} | |
| 93 | - | conn := &clientConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}} | |
| 93 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}}} | |
| 94 | 94 | conn.preambleOnce.Do(func() {}) | |
| 95 | 95 | _, err := conn.ReceiveServer() | |
| 96 | 96 | var failure *protocol.DecodeFailure | |
| @@ -99,7 +99,7 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 99 | 99 | }) | |
| 100 | 100 | t.Run("client receive output data ceiling", func(t *testing.T) { | |
| 101 | 101 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: smallOutput}}} | |
| 102 | - | conn := &clientConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: wire.AbsoluteEnvelopeLimit, outputDataLimit: 4}} | |
| 102 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: wire.AbsoluteEnvelopeLimit, outputDataLimit: 4}}} | |
| 103 | 103 | conn.preambleOnce.Do(func() {}) | |
| 104 | 104 | _, err := conn.ReceiveServer() | |
| 105 | 105 | var failure *protocol.DecodeFailure | |
| @@ -108,14 +108,14 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 108 | 108 | }) | |
| 109 | 109 | t.Run("within ceilings decodes", func(t *testing.T) { | |
| 110 | 110 | serverRaw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeInput}}} | |
| 111 | - | server := &serverConnection{raw: serverRaw, ceilings: defaultProtoCeilings()} | |
| 111 | + | server := &serverConnection{handshake: handshake{raw: serverRaw, ceilings: defaultProtoCeilings()}} | |
| 112 | 112 | server.preambleOnce.Do(func() {}) | |
| 113 | 113 | got, err := server.ReceiveClient() | |
| 114 | 114 | require.NoError(t, err) | |
| 115 | 115 | require.Equal(t, []byte(bytes.Repeat([]byte("x"), 64)), got.(protocol.Input).Data) | |
| 116 | 116 | ||
| 117 | 117 | clientRaw := &scriptedTransport{recv: []wire.Envelope{{Payload: smallOutput}}} | |
| 118 | - | client := &clientConnection{raw: clientRaw, ceilings: defaultProtoCeilings()} | |
| 118 | + | client := &clientConnection{handshake: handshake{raw: clientRaw, ceilings: defaultProtoCeilings()}} | |
| 119 | 119 | client.preambleOnce.Do(func() {}) | |
| 120 | 120 | decoded, err := client.ReceiveServer() | |
| 121 | 121 | require.NoError(t, err) | |
| @@ -127,12 +127,12 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||
| 127 | 127 | // path and advertised capabilities reflect the negotiated ceilings. | |
| 128 | 128 | func TestClientSendAndCapabilitiesUseNegotiatedCeilings(t *testing.T) { | |
| 129 | 129 | raw := &scriptedTransport{} | |
| 130 | - | conn := &clientConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 8, outputDataLimit: 1234}} | |
| 130 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 8, outputDataLimit: 1234}}} | |
| 131 | 131 | conn.preambleOnce.Do(func() {}) | |
| 132 | 132 | require.ErrorIs(t, conn.SendClient(protocol.Input{InputSeq: 1, Data: []byte("hello")}), wire.ErrScanLength) | |
| 133 | 133 | require.Zero(t, raw.sentLen()) | |
| 134 | 134 | require.Equal(t, 1234, conn.Capabilities().OutputDataLimit) | |
| 135 | 135 | ||
| 136 | - | server := &serverConnection{raw: &scriptedTransport{}, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 1, outputDataLimit: 4321}} | |
| 136 | + | server := &serverConnection{handshake: handshake{raw: &scriptedTransport{}, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 1, outputDataLimit: 4321}}} | |
| 137 | 137 | require.Equal(t, 4321, server.Capabilities().OutputDataLimit) | |
| 138 | 138 | } | |
| @@ -42,7 +42,7 @@ func TestServerSendEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 42 | 42 | ||||
| 43 | t.Run("normal send", func(t *testing.T) { | 43 | t.Run("normal send", func(t *testing.T) { | ||
| 44 | raw := &scriptedTransport{} | 44 | raw := &scriptedTransport{} | ||
| 45 | - | conn := &serverConnection{raw: raw, ceilings: envelopeBound} | 45 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: envelopeBound}} |
| 46 | require.ErrorIs(t, conn.SendServer(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | 46 | require.ErrorIs(t, conn.SendServer(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | ||
| 47 | require.Zero(t, raw.sentLen()) | 47 | require.Zero(t, raw.sentLen()) | ||
| 48 | conn.ceilings = outputBound | 48 | conn.ceilings = outputBound | ||
| @@ -51,7 +51,7 @@ func TestServerSendEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 51 | }) | 51 | }) | ||
| 52 | t.Run("async and owned synchronous send", func(t *testing.T) { | 52 | t.Run("async and owned synchronous send", func(t *testing.T) { | ||
| 53 | raw := &asyncScriptedTransport{scriptedTransport: &scriptedTransport{}} | 53 | raw := &asyncScriptedTransport{scriptedTransport: &scriptedTransport{}} | ||
| 54 | - | conn := &serverConnection{raw: raw, ceilings: envelopeBound} | 54 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: envelopeBound}} |
| 55 | require.ErrorIs(t, conn.SendServerAsync(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | 55 | require.ErrorIs(t, conn.SendServerAsync(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | ||
| 56 | require.ErrorIs(t, conn.SendServerSynchronous(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | 56 | require.ErrorIs(t, conn.SendServerSynchronous(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "oversized"}), wire.ErrScanLength) | ||
| 57 | require.Empty(t, raw.async) | 57 | require.Empty(t, raw.async) | ||
| @@ -64,7 +64,7 @@ func TestServerSendEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 64 | }) | 64 | }) | ||
| 65 | t.Run("within ceilings", func(t *testing.T) { | 65 | t.Run("within ceilings", func(t *testing.T) { | ||
| 66 | raw := &scriptedTransport{} | 66 | raw := &scriptedTransport{} | ||
| 67 | - | conn := &serverConnection{raw: raw, ceilings: defaultProtoCeilings()} | 67 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: defaultProtoCeilings()}} |
| 68 | require.NoError(t, conn.SendServer(protocol.Pong{})) | 68 | require.NoError(t, conn.SendServer(protocol.Pong{})) | ||
| 69 | require.NoError(t, conn.SendOutput(testOutput([]byte("12345")))) | 69 | require.NoError(t, conn.SendOutput(testOutput([]byte("12345")))) | ||
| 70 | require.Equal(t, 2, raw.sentLen()) | 70 | require.Equal(t, 2, raw.sentLen()) | ||
| @@ -81,7 +81,7 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 81 | 81 | ||||
| 82 | t.Run("server receive envelope ceiling", func(t *testing.T) { | 82 | t.Run("server receive envelope ceiling", func(t *testing.T) { | ||
| 83 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeInput}}} | 83 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeInput}}} | ||
| 84 | - | conn := &serverConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}} | 84 | + | conn := &serverConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}}} |
| 85 | conn.preambleOnce.Do(func() {}) | 85 | conn.preambleOnce.Do(func() {}) | ||
| 86 | _, err := conn.ReceiveClient() | 86 | _, err := conn.ReceiveClient() | ||
| 87 | var failure *protocol.DecodeFailure | 87 | var failure *protocol.DecodeFailure | ||
| @@ -90,7 +90,7 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 90 | }) | 90 | }) | ||
| 91 | t.Run("client receive envelope ceiling", func(t *testing.T) { | 91 | t.Run("client receive envelope ceiling", func(t *testing.T) { | ||
| 92 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeOutput}}} | 92 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeOutput}}} | ||
| 93 | - | conn := &clientConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}} | 93 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 16, outputDataLimit: uint64(protocol.MaxOutputDataLen)}}} |
| 94 | conn.preambleOnce.Do(func() {}) | 94 | conn.preambleOnce.Do(func() {}) | ||
| 95 | _, err := conn.ReceiveServer() | 95 | _, err := conn.ReceiveServer() | ||
| 96 | var failure *protocol.DecodeFailure | 96 | var failure *protocol.DecodeFailure | ||
| @@ -99,7 +99,7 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 99 | }) | 99 | }) | ||
| 100 | t.Run("client receive output data ceiling", func(t *testing.T) { | 100 | t.Run("client receive output data ceiling", func(t *testing.T) { | ||
| 101 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: smallOutput}}} | 101 | raw := &scriptedTransport{recv: []wire.Envelope{{Payload: smallOutput}}} | ||
| 102 | - | conn := &clientConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: wire.AbsoluteEnvelopeLimit, outputDataLimit: 4}} | 102 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: wire.AbsoluteEnvelopeLimit, outputDataLimit: 4}}} |
| 103 | conn.preambleOnce.Do(func() {}) | 103 | conn.preambleOnce.Do(func() {}) | ||
| 104 | _, err := conn.ReceiveServer() | 104 | _, err := conn.ReceiveServer() | ||
| 105 | var failure *protocol.DecodeFailure | 105 | var failure *protocol.DecodeFailure | ||
| @@ -108,14 +108,14 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 108 | }) | 108 | }) | ||
| 109 | t.Run("within ceilings decodes", func(t *testing.T) { | 109 | t.Run("within ceilings decodes", func(t *testing.T) { | ||
| 110 | serverRaw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeInput}}} | 110 | serverRaw := &scriptedTransport{recv: []wire.Envelope{{Payload: largeInput}}} | ||
| 111 | - | server := &serverConnection{raw: serverRaw, ceilings: defaultProtoCeilings()} | 111 | + | server := &serverConnection{handshake: handshake{raw: serverRaw, ceilings: defaultProtoCeilings()}} |
| 112 | server.preambleOnce.Do(func() {}) | 112 | server.preambleOnce.Do(func() {}) | ||
| 113 | got, err := server.ReceiveClient() | 113 | got, err := server.ReceiveClient() | ||
| 114 | require.NoError(t, err) | 114 | require.NoError(t, err) | ||
| 115 | require.Equal(t, []byte(bytes.Repeat([]byte("x"), 64)), got.(protocol.Input).Data) | 115 | require.Equal(t, []byte(bytes.Repeat([]byte("x"), 64)), got.(protocol.Input).Data) | ||
| 116 | 116 | ||||
| 117 | clientRaw := &scriptedTransport{recv: []wire.Envelope{{Payload: smallOutput}}} | 117 | clientRaw := &scriptedTransport{recv: []wire.Envelope{{Payload: smallOutput}}} | ||
| 118 | - | client := &clientConnection{raw: clientRaw, ceilings: defaultProtoCeilings()} | 118 | + | client := &clientConnection{handshake: handshake{raw: clientRaw, ceilings: defaultProtoCeilings()}} |
| 119 | client.preambleOnce.Do(func() {}) | 119 | client.preambleOnce.Do(func() {}) | ||
| 120 | decoded, err := client.ReceiveServer() | 120 | decoded, err := client.ReceiveServer() | ||
| 121 | require.NoError(t, err) | 121 | require.NoError(t, err) | ||
| @@ -127,12 +127,12 @@ func TestTypedReceiveEnforcesNegotiatedCeilings(t *testing.T) { | |||||
| 127 | // path and advertised capabilities reflect the negotiated ceilings. | 127 | // path and advertised capabilities reflect the negotiated ceilings. | ||
| 128 | func TestClientSendAndCapabilitiesUseNegotiatedCeilings(t *testing.T) { | 128 | func TestClientSendAndCapabilitiesUseNegotiatedCeilings(t *testing.T) { | ||
| 129 | raw := &scriptedTransport{} | 129 | raw := &scriptedTransport{} | ||
| 130 | - | conn := &clientConnection{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 8, outputDataLimit: 1234}} | 130 | + | conn := &clientConnection{handshake: handshake{raw: raw, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 8, outputDataLimit: 1234}}} |
| 131 | conn.preambleOnce.Do(func() {}) | 131 | conn.preambleOnce.Do(func() {}) | ||
| 132 | require.ErrorIs(t, conn.SendClient(protocol.Input{InputSeq: 1, Data: []byte("hello")}), wire.ErrScanLength) | 132 | require.ErrorIs(t, conn.SendClient(protocol.Input{InputSeq: 1, Data: []byte("hello")}), wire.ErrScanLength) | ||
| 133 | require.Zero(t, raw.sentLen()) | 133 | require.Zero(t, raw.sentLen()) | ||
| 134 | require.Equal(t, 1234, conn.Capabilities().OutputDataLimit) | 134 | require.Equal(t, 1234, conn.Capabilities().OutputDataLimit) | ||
| 135 | 135 | ||||
| 136 | - | server := &serverConnection{raw: &scriptedTransport{}, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 1, outputDataLimit: 4321}} | 136 | + | server := &serverConnection{handshake: handshake{raw: &scriptedTransport{}, ceilings: protoCeilings{maxReceiveEnvelopeBytes: 1, outputDataLimit: 4321}}} |
| 137 | require.Equal(t, 4321, server.Capabilities().OutputDataLimit) | 137 | require.Equal(t, 4321, server.Capabilities().OutputDataLimit) | ||
| 138 | } | 138 | } | ||
M internal/adapters/sessionwire/server.go +7 -124
| @@ -2,9 +2,7 @@ | |||
| 2 | 2 | package sessionwire | |
| 3 | 3 | ||
| 4 | 4 | import ( | |
| 5 | - | "context" | |
| 6 | 5 | "errors" | |
| 7 | - | "sync" | |
| 8 | 6 | "time" | |
| 9 | 7 | ||
| 10 | 8 | "google.golang.org/protobuf/encoding/protowire" | |
| @@ -22,35 +20,15 @@ var ( | |||
| 22 | 20 | ) | |
| 23 | 21 | ||
| 24 | 22 | type serverConnection struct { | |
| 25 | - | raw wire.Transport | |
| 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 | |
| 32 | - | ||
| 33 | - | preambleOnce sync.Once | |
| 34 | - | preambleErr error | |
| 35 | - | deadline time.Time | |
| 23 | + | handshake | |
| 36 | 24 | ||
| 37 | 25 | // admission is the provisioned admission metadata the accepting side | |
| 38 | - | // stamped, or the zero value with hasAdmission=false for a connection built | |
| 39 | - | // by a legacy constructor that carried no admission. It is never mutated | |
| 40 | - | // after construction and is exposed only through SessionAdmission, which | |
| 41 | - | // returns a deep copy. | |
| 26 | + | // stamped, or the zero value with hasAdmission=false for a connection whose | |
| 27 | + | // accepting side provisioned none (a direct local daemon socket). It is | |
| 28 | + | // never mutated after construction and is exposed only through | |
| 29 | + | // SessionAdmission, which returns a deep copy. | |
| 42 | 30 | admission ports.SessionAdmission | |
| 43 | 31 | hasAdmission bool | |
| 44 | - | ||
| 45 | - | // preambleDone closes exactly once when the handshake has run, whether it | |
| 46 | - | // succeeded or failed. It backs the handshake completion seam a mux listener | |
| 47 | - | // exposes to the daemon; the absolute deadline is never restarted. | |
| 48 | - | preambleDone chan struct{} | |
| 49 | - | ||
| 50 | - | hooksMu sync.Mutex | |
| 51 | - | handshakeDone bool | |
| 52 | - | hooksRun bool | |
| 53 | - | hooks []func() | |
| 54 | 32 | } | |
| 55 | 33 | ||
| 56 | 34 | var ( | |
| @@ -114,12 +92,9 @@ func newServerConnection(raw wire.Transport, deadline time.Time, admission ports | |||
| 114 | 92 | deadline = time.Now().Add(protocol.HandshakeTimeout) | |
| 115 | 93 | } | |
| 116 | 94 | return &serverConnection{ | |
| 117 | - | raw: raw, | |
| 118 | - | ceilings: defaultProtoCeilings(), | |
| 119 | - | deadline: deadline, | |
| 95 | + | handshake: newHandshake(raw, deadline), | |
| 120 | 96 | admission: admission.Clone(), | |
| 121 | 97 | hasAdmission: hasAdmission, | |
| 122 | - | preambleDone: make(chan struct{}), | |
| 123 | 98 | } | |
| 124 | 99 | } | |
| 125 | 100 | ||
| @@ -135,100 +110,8 @@ func (c *serverConnection) SessionAdmission() (ports.SessionAdmission, bool) { | |||
| 135 | 110 | return c.admission.Clone(), true | |
| 136 | 111 | } | |
| 137 | 112 | ||
| 138 | - | // HandshakeDeadline returns the absolute local deadline of the session | |
| 139 | - | // handshake this connection accepted. It is fixed for the connection's | |
| 140 | - | // lifetime and is never restarted: the preamble, Hello/Welcome, and the first | |
| 141 | - | // committed publication all share it. | |
| 142 | - | func (c *serverConnection) HandshakeDeadline() time.Time { return c.deadline } | |
| 143 | - | ||
| 144 | - | // HandshakeDone returns a channel that is closed exactly once when the | |
| 145 | - | // handshake has run, whether it succeeded or failed. | |
| 146 | - | func (c *serverConnection) HandshakeDone() <-chan struct{} { return c.preambleDone } | |
| 147 | - | ||
| 148 | - | // OnHandshakeComplete registers fn to run exactly once when the handshake has | |
| 149 | - | // run. Registering after completion runs fn immediately; a nil fn is ignored. | |
| 150 | - | func (c *serverConnection) OnHandshakeComplete(fn func()) { | |
| 151 | - | if fn == nil { | |
| 152 | - | return | |
| 153 | - | } | |
| 154 | - | c.hooksMu.Lock() | |
| 155 | - | if c.handshakeDone { | |
| 156 | - | c.hooksMu.Unlock() | |
| 157 | - | fn() | |
| 158 | - | return | |
| 159 | - | } | |
| 160 | - | c.hooks = append(c.hooks, fn) | |
| 161 | - | c.hooksMu.Unlock() | |
| 162 | - | } | |
| 163 | - | ||
| 164 | - | // finishPreamble publishes handshake completion from inside the preamble's | |
| 165 | - | // sync.Once: it closes the done channel and marks the handshake complete so a | |
| 166 | - | // later registration runs immediately. It deliberately runs no registered hook, | |
| 167 | - | // because a hook that calls back into the connection would re-enter sync.Once | |
| 168 | - | // and deadlock; runHandshakeHooks runs the hooks after the Once body returns. | |
| 169 | - | // It tolerates a zero-value connection that never allocated the done channel | |
| 170 | - | // (test literals and seam-only construction), so closing is nil-safe. | |
| 171 | - | func (c *serverConnection) finishPreamble() { | |
| 172 | - | if c.preambleDone != nil { | |
| 173 | - | close(c.preambleDone) | |
| 174 | - | } | |
| 175 | - | c.hooksMu.Lock() | |
| 176 | - | c.handshakeDone = true | |
| 177 | - | c.hooksMu.Unlock() | |
| 178 | - | } | |
| 179 | - | ||
| 180 | - | // runHandshakeHooks runs the completion hooks registered before the handshake | |
| 181 | - | // finished exactly once, after the preamble's sync.Once body has returned. | |
| 182 | - | // Running them outside the Once is what lets a hook call back into the | |
| 183 | - | // connection - for example to receive a message - without re-entering | |
| 184 | - | // sync.Once and deadlocking. Every ensurePreamble caller invokes it; only the | |
| 185 | - | // first drains the hooks, and a registrant that arrives after completion | |
| 186 | - | // observes the completed flag and runs itself. | |
| 187 | - | func (c *serverConnection) runHandshakeHooks() { | |
| 188 | - | c.hooksMu.Lock() | |
| 189 | - | if c.hooksRun { | |
| 190 | - | c.hooksMu.Unlock() | |
| 191 | - | return | |
| 192 | - | } | |
| 193 | - | c.hooksRun = true | |
| 194 | - | hooks := c.hooks | |
| 195 | - | c.hooks = nil | |
| 196 | - | c.hooksMu.Unlock() | |
| 197 | - | for _, fn := range hooks { | |
| 198 | - | fn() | |
| 199 | - | } | |
| 200 | - | } | |
| 201 | - | ||
| 202 | 113 | func (c *serverConnection) ensurePreamble() error { | |
| 203 | - | c.preambleOnce.Do(func() { | |
| 204 | - | ctx, cancel := context.WithDeadline(context.Background(), c.deadline) | |
| 205 | - | defer cancel() | |
| 206 | - | next, err := runProtoServerPreamble(ctx, c.raw, c.limits()) | |
| 207 | - | c.publishLimits(next) | |
| 208 | - | c.preambleErr = err | |
| 209 | - | if err != nil { | |
| 210 | - | _ = c.raw.Close() | |
| 211 | - | } | |
| 212 | - | c.finishPreamble() | |
| 213 | - | }) | |
| 214 | - | c.runHandshakeHooks() | |
| 215 | - | return c.preambleErr | |
| 216 | - | } | |
| 217 | - | ||
| 218 | - | // limits returns the negotiated ceilings under the lock the preamble publishes | |
| 219 | - | // them through. Send paths may read them without the lock once ensurePreamble | |
| 220 | - | // has returned, because that call orders the publish before the read. | |
| 221 | - | func (c *serverConnection) limits() protoCeilings { | |
| 222 | - | c.ceilingsMu.Lock() | |
| 223 | - | defer c.ceilingsMu.Unlock() | |
| 224 | - | return c.ceilings | |
| 225 | - | } | |
| 226 | - | ||
| 227 | - | // publishLimits records the ceilings one preamble negotiation produced. | |
| 228 | - | func (c *serverConnection) publishLimits(next protoCeilings) { | |
| 229 | - | c.ceilingsMu.Lock() | |
| 230 | - | c.ceilings = next | |
| 231 | - | c.ceilingsMu.Unlock() | |
| 114 | + | return c.handshake.ensurePreamble(runProtoServerPreamble) | |
| 232 | 115 | } | |
| 233 | 116 | ||
| 234 | 117 | func (c *serverConnection) ReceiveClient() (protocol.ClientMessage, error) { | |
| @@ -2,9 +2,7 @@ | |||||
| 2 | package sessionwire | 2 | package sessionwire | ||
| 3 | 3 | ||||
| 4 | import ( | 4 | import ( | ||
| 5 | - | "context" | |||
| 6 | "errors" | 5 | "errors" | ||
| 7 | - | "sync" | |||
| 8 | "time" | 6 | "time" | ||
| 9 | 7 | ||||
| 10 | "google.golang.org/protobuf/encoding/protowire" | 8 | "google.golang.org/protobuf/encoding/protowire" | ||
| @@ -22,35 +20,15 @@ var ( | |||||
| 22 | ) | 20 | ) | ||
| 23 | 21 | ||||
| 24 | type serverConnection struct { | 22 | type serverConnection struct { | ||
| 25 | - | raw wire.Transport | 23 | + | handshake |
| 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 | |||
| 32 | - | ||||
| 33 | - | preambleOnce sync.Once | |||
| 34 | - | preambleErr error | |||
| 35 | - | deadline time.Time | |||
| 36 | 24 | ||||
| 37 | // admission is the provisioned admission metadata the accepting side | 25 | // admission is the provisioned admission metadata the accepting side | ||
| 38 | - | // stamped, or the zero value with hasAdmission=false for a connection built | 26 | + | // stamped, or the zero value with hasAdmission=false for a connection whose |
| 39 | - | // by a legacy constructor that carried no admission. It is never mutated | 27 | + | // accepting side provisioned none (a direct local daemon socket). It is |
| 40 | - | // after construction and is exposed only through SessionAdmission, which | 28 | + | // never mutated after construction and is exposed only through |
| 41 | - | // returns a deep copy. | 29 | + | // SessionAdmission, which returns a deep copy. |
| 42 | admission ports.SessionAdmission | 30 | admission ports.SessionAdmission | ||
| 43 | hasAdmission bool | 31 | hasAdmission bool | ||
| 44 | - | ||||
| 45 | - | // preambleDone closes exactly once when the handshake has run, whether it | |||
| 46 | - | // succeeded or failed. It backs the handshake completion seam a mux listener | |||
| 47 | - | // exposes to the daemon; the absolute deadline is never restarted. | |||
| 48 | - | preambleDone chan struct{} | |||
| 49 | - | ||||
| 50 | - | hooksMu sync.Mutex | |||
| 51 | - | handshakeDone bool | |||
| 52 | - | hooksRun bool | |||
| 53 | - | hooks []func() | |||
| 54 | } | 32 | } | ||
| 55 | 33 | ||||
| 56 | var ( | 34 | var ( | ||
| @@ -114,12 +92,9 @@ func newServerConnection(raw wire.Transport, deadline time.Time, admission ports | |||||
| 114 | deadline = time.Now().Add(protocol.HandshakeTimeout) | 92 | deadline = time.Now().Add(protocol.HandshakeTimeout) | ||
| 115 | } | 93 | } | ||
| 116 | return &serverConnection{ | 94 | return &serverConnection{ | ||
| 117 | - | raw: raw, | 95 | + | handshake: newHandshake(raw, deadline), |
| 118 | - | ceilings: defaultProtoCeilings(), | |||
| 119 | - | deadline: deadline, | |||
| 120 | admission: admission.Clone(), | 96 | admission: admission.Clone(), | ||
| 121 | hasAdmission: hasAdmission, | 97 | hasAdmission: hasAdmission, | ||
| 122 | - | preambleDone: make(chan struct{}), | |||
| 123 | } | 98 | } | ||
| 124 | } | 99 | } | ||
| 125 | 100 | ||||
| @@ -135,100 +110,8 @@ func (c *serverConnection) SessionAdmission() (ports.SessionAdmission, bool) { | |||||
| 135 | return c.admission.Clone(), true | 110 | return c.admission.Clone(), true | ||
| 136 | } | 111 | } | ||
| 137 | 112 | ||||
| 138 | - | // HandshakeDeadline returns the absolute local deadline of the session | |||
| 139 | - | // handshake this connection accepted. It is fixed for the connection's | |||
| 140 | - | // lifetime and is never restarted: the preamble, Hello/Welcome, and the first | |||
| 141 | - | // committed publication all share it. | |||
| 142 | - | func (c *serverConnection) HandshakeDeadline() time.Time { return c.deadline } | |||
| 143 | - | ||||
| 144 | - | // HandshakeDone returns a channel that is closed exactly once when the | |||
| 145 | - | // handshake has run, whether it succeeded or failed. | |||
| 146 | - | func (c *serverConnection) HandshakeDone() <-chan struct{} { return c.preambleDone } | |||
| 147 | - | ||||
| 148 | - | // OnHandshakeComplete registers fn to run exactly once when the handshake has | |||
| 149 | - | // run. Registering after completion runs fn immediately; a nil fn is ignored. | |||
| 150 | - | func (c *serverConnection) OnHandshakeComplete(fn func()) { | |||
| 151 | - | if fn == nil { | |||
| 152 | - | return | |||
| 153 | - | } | |||
| 154 | - | c.hooksMu.Lock() | |||
| 155 | - | if c.handshakeDone { | |||
| 156 | - | c.hooksMu.Unlock() | |||
| 157 | - | fn() | |||
| 158 | - | return | |||
| 159 | - | } | |||
| 160 | - | c.hooks = append(c.hooks, fn) | |||
| 161 | - | c.hooksMu.Unlock() | |||
| 162 | - | } | |||
| 163 | - | ||||
| 164 | - | // finishPreamble publishes handshake completion from inside the preamble's | |||
| 165 | - | // sync.Once: it closes the done channel and marks the handshake complete so a | |||
| 166 | - | // later registration runs immediately. It deliberately runs no registered hook, | |||
| 167 | - | // because a hook that calls back into the connection would re-enter sync.Once | |||
| 168 | - | // and deadlock; runHandshakeHooks runs the hooks after the Once body returns. | |||
| 169 | - | // It tolerates a zero-value connection that never allocated the done channel | |||
| 170 | - | // (test literals and seam-only construction), so closing is nil-safe. | |||
| 171 | - | func (c *serverConnection) finishPreamble() { | |||
| 172 | - | if c.preambleDone != nil { | |||
| 173 | - | close(c.preambleDone) | |||
| 174 | - | } | |||
| 175 | - | c.hooksMu.Lock() | |||
| 176 | - | c.handshakeDone = true | |||
| 177 | - | c.hooksMu.Unlock() | |||
| 178 | - | } | |||
| 179 | - | ||||
| 180 | - | // runHandshakeHooks runs the completion hooks registered before the handshake | |||
| 181 | - | // finished exactly once, after the preamble's sync.Once body has returned. | |||
| 182 | - | // Running them outside the Once is what lets a hook call back into the | |||
| 183 | - | // connection - for example to receive a message - without re-entering | |||
| 184 | - | // sync.Once and deadlocking. Every ensurePreamble caller invokes it; only the | |||
| 185 | - | // first drains the hooks, and a registrant that arrives after completion | |||
| 186 | - | // observes the completed flag and runs itself. | |||
| 187 | - | func (c *serverConnection) runHandshakeHooks() { | |||
| 188 | - | c.hooksMu.Lock() | |||
| 189 | - | if c.hooksRun { | |||
| 190 | - | c.hooksMu.Unlock() | |||
| 191 | - | return | |||
| 192 | - | } | |||
| 193 | - | c.hooksRun = true | |||
| 194 | - | hooks := c.hooks | |||
| 195 | - | c.hooks = nil | |||
| 196 | - | c.hooksMu.Unlock() | |||
| 197 | - | for _, fn := range hooks { | |||
| 198 | - | fn() | |||
| 199 | - | } | |||
| 200 | - | } | |||
| 201 | - | ||||
| 202 | func (c *serverConnection) ensurePreamble() error { | 113 | func (c *serverConnection) ensurePreamble() error { | ||
| 203 | - | c.preambleOnce.Do(func() { | 114 | + | return c.handshake.ensurePreamble(runProtoServerPreamble) |
| 204 | - | ctx, cancel := context.WithDeadline(context.Background(), c.deadline) | |||
| 205 | - | defer cancel() | |||
| 206 | - | next, err := runProtoServerPreamble(ctx, c.raw, c.limits()) | |||
| 207 | - | c.publishLimits(next) | |||
| 208 | - | c.preambleErr = err | |||
| 209 | - | if err != nil { | |||
| 210 | - | _ = c.raw.Close() | |||
| 211 | - | } | |||
| 212 | - | c.finishPreamble() | |||
| 213 | - | }) | |||
| 214 | - | c.runHandshakeHooks() | |||
| 215 | - | return c.preambleErr | |||
| 216 | - | } | |||
| 217 | - | ||||
| 218 | - | // limits returns the negotiated ceilings under the lock the preamble publishes | |||
| 219 | - | // them through. Send paths may read them without the lock once ensurePreamble | |||
| 220 | - | // has returned, because that call orders the publish before the read. | |||
| 221 | - | func (c *serverConnection) limits() protoCeilings { | |||
| 222 | - | c.ceilingsMu.Lock() | |||
| 223 | - | defer c.ceilingsMu.Unlock() | |||
| 224 | - | return c.ceilings | |||
| 225 | - | } | |||
| 226 | - | ||||
| 227 | - | // publishLimits records the ceilings one preamble negotiation produced. | |||
| 228 | - | func (c *serverConnection) publishLimits(next protoCeilings) { | |||
| 229 | - | c.ceilingsMu.Lock() | |||
| 230 | - | c.ceilings = next | |||
| 231 | - | c.ceilingsMu.Unlock() | |||
| 232 | } | 115 | } | ||
| 233 | 116 | ||||
| 234 | func (c *serverConnection) ReceiveClient() (protocol.ClientMessage, error) { | 117 | func (c *serverConnection) ReceiveClient() (protocol.ClientMessage, error) { | ||