detached fecc0c5

fix(remote): unblock same-peer switch control

commit
fecc0c58b7569ca3a8af3f687e418352a5be36b7
author
brice · · 2026-08-23 12:51 UTC
parent
8b24e21
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
6 files · +179 -41
Unified
UnifiedSplit
M internal/ports/same_peer_switch_test.go +51 -28
@@ -15,36 +15,59 @@ func samePeerSwitchTarget() ExactSessionTarget {
1515
1616 func TestSamePeerSwitchWireStrict(t *testing.T) {
1717 request := SamePeerSwitchRequest{RequestID: 7, Target: samePeerSwitchTarget(), PreferredTabID: "tab-2"}
18- payload, err := MarshalSamePeerSwitchRequest(request)
19- if err != nil {
20- t.Fatal(err)
21- }
22- want := append([]byte{0, 0, 0, 0, 0, 0, 0, 7, 9}, make([]byte, 15)...)
23- want = append(want, 0, 4, 'w', 'o', 'r', 'k', 0, 5, 't', 'a', 'b', '-', '2')
24- if !bytes.Equal(payload, want) {
25- t.Fatalf("request payload = %x, want %x", payload, want)
26- }
27- decoded, err := UnmarshalSamePeerSwitchRequest(payload)
28- if err != nil || decoded != request {
29- t.Fatalf("request = %+v, error = %v, want %+v", decoded, err, request)
30- }
31- assertAllPrefixesFail(t, payload, UnmarshalSamePeerSwitchRequest)
32- assertTrailingGarbageFails(t, payload, UnmarshalSamePeerSwitchRequest)
33-
18+ requestWire := append([]byte{0, 0, 0, 0, 0, 0, 0, 7, 9}, make([]byte, 15)...)
19+ requestWire = append(requestWire, 0, 4, 'w', 'o', 'r', 'k', 0, 5, 't', 'a', 'b', '-', '2')
3420 failure := SamePeerSwitchFailure{RequestID: 7, Code: SamePeerSwitchStaleTarget}
35- failurePayload, err := MarshalSamePeerSwitchFailure(failure)
36- if err != nil {
37- t.Fatal(err)
38- }
39- if want := []byte{0, 0, 0, 0, 0, 0, 0, 7, 1}; !bytes.Equal(failurePayload, want) {
40- t.Fatalf("failure payload = %x, want %x", failurePayload, want)
41- }
42- decodedFailure, err := UnmarshalSamePeerSwitchFailure(failurePayload)
43- if err != nil || decodedFailure != failure {
44- t.Fatalf("failure = %+v, error = %v, want %+v", decodedFailure, err, failure)
21+
22+ for _, tt := range []struct {
23+ name string
24+ want []byte
25+ marshal func() ([]byte, error)
26+ decode func([]byte) (any, error)
27+ assert func(*testing.T, any)
28+ }{
29+ {
30+ name: "request",
31+ want: requestWire,
32+ marshal: func() ([]byte, error) { return MarshalSamePeerSwitchRequest(request) },
33+ decode: func(payload []byte) (any, error) { return UnmarshalSamePeerSwitchRequest(payload) },
34+ assert: func(t *testing.T, decoded any) {
35+ t.Helper()
36+ if got := decoded.(SamePeerSwitchRequest); got != request {
37+ t.Fatalf("request = %+v, want %+v", got, request)
38+ }
39+ },
40+ },
41+ {
42+ name: "failure",
43+ want: []byte{0, 0, 0, 0, 0, 0, 0, 7, 1},
44+ marshal: func() ([]byte, error) { return MarshalSamePeerSwitchFailure(failure) },
45+ decode: func(payload []byte) (any, error) { return UnmarshalSamePeerSwitchFailure(payload) },
46+ assert: func(t *testing.T, decoded any) {
47+ t.Helper()
48+ if got := decoded.(SamePeerSwitchFailure); got != failure {
49+ t.Fatalf("failure = %+v, want %+v", got, failure)
50+ }
51+ },
52+ },
53+ } {
54+ t.Run(tt.name, func(t *testing.T) {
55+ payload, err := tt.marshal()
56+ if err != nil {
57+ t.Fatal(err)
58+ }
59+ if !bytes.Equal(payload, tt.want) {
60+ t.Fatalf("payload = %x, want %x", payload, tt.want)
61+ }
62+ decoded, err := tt.decode(payload)
63+ if err != nil {
64+ t.Fatal(err)
65+ }
66+ tt.assert(t, decoded)
67+ assertAllPrefixesFail(t, payload, tt.decode)
68+ assertTrailingGarbageFails(t, payload, tt.decode)
69+ })
4570 }
46- assertAllPrefixesFail(t, failurePayload, UnmarshalSamePeerSwitchFailure)
47- assertTrailingGarbageFails(t, failurePayload, UnmarshalSamePeerSwitchFailure)
4871 }
4972
5073 func TestSamePeerSwitchWireRejectsInvalidValues(t *testing.T) {
@@ -15,36 +15,59 @@ func samePeerSwitchTarget() ExactSessionTarget {
15 15
16 func TestSamePeerSwitchWireStrict(t *testing.T) {16 func TestSamePeerSwitchWireStrict(t *testing.T) {
17 request := SamePeerSwitchRequest{RequestID: 7, Target: samePeerSwitchTarget(), PreferredTabID: "tab-2"}17 request := SamePeerSwitchRequest{RequestID: 7, Target: samePeerSwitchTarget(), PreferredTabID: "tab-2"}
18- payload, err := MarshalSamePeerSwitchRequest(request)18+ requestWire := append([]byte{0, 0, 0, 0, 0, 0, 0, 7, 9}, make([]byte, 15)...)
19- if err != nil {19+ requestWire = append(requestWire, 0, 4, 'w', 'o', 'r', 'k', 0, 5, 't', 'a', 'b', '-', '2')
20- t.Fatal(err)
21- }
22- want := append([]byte{0, 0, 0, 0, 0, 0, 0, 7, 9}, make([]byte, 15)...)
23- want = append(want, 0, 4, 'w', 'o', 'r', 'k', 0, 5, 't', 'a', 'b', '-', '2')
24- if !bytes.Equal(payload, want) {
25- t.Fatalf("request payload = %x, want %x", payload, want)
26- }
27- decoded, err := UnmarshalSamePeerSwitchRequest(payload)
28- if err != nil || decoded != request {
29- t.Fatalf("request = %+v, error = %v, want %+v", decoded, err, request)
30- }
31- assertAllPrefixesFail(t, payload, UnmarshalSamePeerSwitchRequest)
32- assertTrailingGarbageFails(t, payload, UnmarshalSamePeerSwitchRequest)
33-
34 failure := SamePeerSwitchFailure{RequestID: 7, Code: SamePeerSwitchStaleTarget}20 failure := SamePeerSwitchFailure{RequestID: 7, Code: SamePeerSwitchStaleTarget}
35- failurePayload, err := MarshalSamePeerSwitchFailure(failure)21+
36- if err != nil {22+ for _, tt := range []struct {
37- t.Fatal(err)23+ name string
38- }24+ want []byte
39- if want := []byte{0, 0, 0, 0, 0, 0, 0, 7, 1}; !bytes.Equal(failurePayload, want) {25+ marshal func() ([]byte, error)
40- t.Fatalf("failure payload = %x, want %x", failurePayload, want)26+ decode func([]byte) (any, error)
41- }27+ assert func(*testing.T, any)
42- decodedFailure, err := UnmarshalSamePeerSwitchFailure(failurePayload)28+ }{
43- if err != nil || decodedFailure != failure {29+ {
44- t.Fatalf("failure = %+v, error = %v, want %+v", decodedFailure, err, failure)30+ name: "request",
31+ want: requestWire,
32+ marshal: func() ([]byte, error) { return MarshalSamePeerSwitchRequest(request) },
33+ decode: func(payload []byte) (any, error) { return UnmarshalSamePeerSwitchRequest(payload) },
34+ assert: func(t *testing.T, decoded any) {
35+ t.Helper()
36+ if got := decoded.(SamePeerSwitchRequest); got != request {
37+ t.Fatalf("request = %+v, want %+v", got, request)
38+ }
39+ },
40+ },
41+ {
42+ name: "failure",
43+ want: []byte{0, 0, 0, 0, 0, 0, 0, 7, 1},
44+ marshal: func() ([]byte, error) { return MarshalSamePeerSwitchFailure(failure) },
45+ decode: func(payload []byte) (any, error) { return UnmarshalSamePeerSwitchFailure(payload) },
46+ assert: func(t *testing.T, decoded any) {
47+ t.Helper()
48+ if got := decoded.(SamePeerSwitchFailure); got != failure {
49+ t.Fatalf("failure = %+v, want %+v", got, failure)
50+ }
51+ },
52+ },
53+ } {
54+ t.Run(tt.name, func(t *testing.T) {
55+ payload, err := tt.marshal()
56+ if err != nil {
57+ t.Fatal(err)
58+ }
59+ if !bytes.Equal(payload, tt.want) {
60+ t.Fatalf("payload = %x, want %x", payload, tt.want)
61+ }
62+ decoded, err := tt.decode(payload)
63+ if err != nil {
64+ t.Fatal(err)
65+ }
66+ tt.assert(t, decoded)
67+ assertAllPrefixesFail(t, payload, tt.decode)
68+ assertTrailingGarbageFails(t, payload, tt.decode)
69+ })
45 }70 }
46- assertAllPrefixesFail(t, failurePayload, UnmarshalSamePeerSwitchFailure)
47- assertTrailingGarbageFails(t, failurePayload, UnmarshalSamePeerSwitchFailure)
48 }71 }
49 72
50 func TestSamePeerSwitchWireRejectsInvalidValues(t *testing.T) {73 func TestSamePeerSwitchWireRejectsInvalidValues(t *testing.T) {
M internal/usecase/client/ack_queue_test.go +53 -1
@@ -5,6 +5,7 @@ import (
55 "io"
66 "log/slog"
77 "testing"
8+ "time"
89
910 "github.com/bnema/vev/internal/ports"
1011 "github.com/stretchr/testify/require"
@@ -31,7 +32,7 @@ func TestCumulativeAckBypassesFullNormalSendQueue(t *testing.T) {
3132 defer cancel()
3233 tr := &ackRecordingTransport{sent: make(chan ports.Frame, sendQueueDepth+1)}
3334 errCh := make(chan error, 1)
34- go runSender(ctx, cancel, tr, normal, nil, acks, errCh, slog.Default())
35+ go runSender(ctx, cancel, tr, nil, normal, nil, acks, errCh, slog.Default())
3536
3637 first := <-tr.sent
3738 require.Equal(t, ports.MsgAck, first.Type)
@@ -41,4 +42,55 @@ func TestCumulativeAckBypassesFullNormalSendQueue(t *testing.T) {
4142 require.Equal(t, uint64(maxUnackedOutputStatesForTest), ack.State)
4243 }
4344
45+func TestSamePeerSwitchControlBypassesHeldInput(t *testing.T) {
46+ ctx, cancel := context.WithCancel(context.Background())
47+ defer cancel()
48+ normal := make(chan ports.Frame, 1)
49+ control := make(chan ports.Frame, 1)
50+ gate := newSamePeerInputGate()
51+ gate.setPaused(true)
52+ held := make(chan struct{})
53+ gate.afterInputHeld = func() { close(held) }
54+ transport := &ackRecordingTransport{sent: make(chan ports.Frame, 2)}
55+ acks := newCumulativeAckQueue()
56+ errs := make(chan error, 1)
57+ go runSender(ctx, cancel, transport, control, normal, gate, acks, errs, slog.Default())
58+
59+ normal <- ports.Frame{Type: ports.MsgInput, Payload: []byte("held")}
60+ awaitSenderSignal(t, held)
61+ control <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: []byte("switch")}
62+ if got := awaitSenderFrame(t, transport.sent); got.Type != ports.MsgSamePeerSwitchRequest {
63+ t.Fatalf("first frame = %d, want same-peer switch request", got.Type)
64+ }
65+ select {
66+ case got := <-transport.sent:
67+ t.Fatalf("held input sent before switch resolution: %d", got.Type)
68+ default:
69+ }
70+ gate.setPaused(false)
71+ if got := awaitSenderFrame(t, transport.sent); got.Type != ports.MsgInput {
72+ t.Fatalf("released frame = %d, want input", got.Type)
73+ }
74+}
75+
76+func awaitSenderSignal(t *testing.T, signal <-chan struct{}) {
77+ t.Helper()
78+ select {
79+ case <-signal:
80+ case <-time.After(time.Second):
81+ t.Fatal("sender did not hold input")
82+ }
83+}
84+
85+func awaitSenderFrame(t *testing.T, frames <-chan ports.Frame) ports.Frame {
86+ t.Helper()
87+ select {
88+ case frame := <-frames:
89+ return frame
90+ case <-time.After(time.Second):
91+ t.Fatal("sender did not emit frame")
92+ return ports.Frame{}
93+ }
94+}
95+
4496 const maxUnackedOutputStatesForTest = 8
@@ -5,6 +5,7 @@ import (
5 "io"5 "io"
6 "log/slog"6 "log/slog"
7 "testing"7 "testing"
8+ "time"
8 9
9 "github.com/bnema/vev/internal/ports"10 "github.com/bnema/vev/internal/ports"
10 "github.com/stretchr/testify/require"11 "github.com/stretchr/testify/require"
@@ -31,7 +32,7 @@ func TestCumulativeAckBypassesFullNormalSendQueue(t *testing.T) {
31 defer cancel()32 defer cancel()
32 tr := &ackRecordingTransport{sent: make(chan ports.Frame, sendQueueDepth+1)}33 tr := &ackRecordingTransport{sent: make(chan ports.Frame, sendQueueDepth+1)}
33 errCh := make(chan error, 1)34 errCh := make(chan error, 1)
34- go runSender(ctx, cancel, tr, normal, nil, acks, errCh, slog.Default())35+ go runSender(ctx, cancel, tr, nil, normal, nil, acks, errCh, slog.Default())
35 36
36 first := <-tr.sent37 first := <-tr.sent
37 require.Equal(t, ports.MsgAck, first.Type)38 require.Equal(t, ports.MsgAck, first.Type)
@@ -41,4 +42,55 @@ func TestCumulativeAckBypassesFullNormalSendQueue(t *testing.T) {
41 require.Equal(t, uint64(maxUnackedOutputStatesForTest), ack.State)42 require.Equal(t, uint64(maxUnackedOutputStatesForTest), ack.State)
42 }43 }
43 44
45+func TestSamePeerSwitchControlBypassesHeldInput(t *testing.T) {
46+ ctx, cancel := context.WithCancel(context.Background())
47+ defer cancel()
48+ normal := make(chan ports.Frame, 1)
49+ control := make(chan ports.Frame, 1)
50+ gate := newSamePeerInputGate()
51+ gate.setPaused(true)
52+ held := make(chan struct{})
53+ gate.afterInputHeld = func() { close(held) }
54+ transport := &ackRecordingTransport{sent: make(chan ports.Frame, 2)}
55+ acks := newCumulativeAckQueue()
56+ errs := make(chan error, 1)
57+ go runSender(ctx, cancel, transport, control, normal, gate, acks, errs, slog.Default())
58+
59+ normal <- ports.Frame{Type: ports.MsgInput, Payload: []byte("held")}
60+ awaitSenderSignal(t, held)
61+ control <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: []byte("switch")}
62+ if got := awaitSenderFrame(t, transport.sent); got.Type != ports.MsgSamePeerSwitchRequest {
63+ t.Fatalf("first frame = %d, want same-peer switch request", got.Type)
64+ }
65+ select {
66+ case got := <-transport.sent:
67+ t.Fatalf("held input sent before switch resolution: %d", got.Type)
68+ default:
69+ }
70+ gate.setPaused(false)
71+ if got := awaitSenderFrame(t, transport.sent); got.Type != ports.MsgInput {
72+ t.Fatalf("released frame = %d, want input", got.Type)
73+ }
74+}
75+
76+func awaitSenderSignal(t *testing.T, signal <-chan struct{}) {
77+ t.Helper()
78+ select {
79+ case <-signal:
80+ case <-time.After(time.Second):
81+ t.Fatal("sender did not hold input")
82+ }
83+}
84+
85+func awaitSenderFrame(t *testing.T, frames <-chan ports.Frame) ports.Frame {
86+ t.Helper()
87+ select {
88+ case frame := <-frames:
89+ return frame
90+ case <-time.After(time.Second):
91+ t.Fatal("sender did not emit frame")
92+ return ports.Frame{}
93+ }
94+}
95+
44 const maxUnackedOutputStatesForTest = 896 const maxUnackedOutputStatesForTest = 8
M internal/usecase/client/client.go +51 -7
@@ -1197,6 +1197,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
11971197 defer revokeInputClaim()
11981198
11991199 sendCh := make(chan ports.Frame, sendQueueDepth)
1200+ controlCh := make(chan ports.Frame, 1)
12001201 inputGate := newSamePeerInputGate()
12011202 ackQueue := newCumulativeAckQueue()
12021203 sendErrCh := make(chan error, 1)
@@ -1338,7 +1339,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
13381339 // or unchanged screen may legitimately produce no initial Output frame.
13391340 endHandshake()
13401341 senderStarted = true
1341- go runSender(loopCtx, cancel, transport, sendCh, inputGate, ackQueue, sendErrCh, log)
1342+ go runSender(loopCtx, cancel, transport, controlCh, sendCh, inputGate, ackQueue, sendErrCh, log)
13421343 stdinDone := make(chan struct{})
13431344 go func() {
13441345 defer close(stdinDone)
@@ -1589,7 +1590,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
15891590 samePeerSwitch = &samePeerSwitchPending{requestID: nextSamePeerRequestID, target: *target.ExactTarget}
15901591 inputGate.setPaused(true)
15911592 select {
1592- case sendCh <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: payload}:
1593+ case controlCh <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: payload}:
15931594 continue
15941595 case <-loopCtx.Done():
15951596 return loopCanceledResult()
@@ -1763,7 +1764,9 @@ func (q *cumulativeAckQueue) take() (uint64, uint64) {
17631764 return epoch, state
17641765 }
17651766
1766-func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, inputGate *samePeerInputGate, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
1767+// runSender preserves normal-frame order while allowing switch control and
1768+// output ACKs to make progress when raw input is held for a same-peer switch.
1769+func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, control, in <-chan ports.Frame, inputGate *samePeerInputGate, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
17671770 defer log.Debug("sender pump exited")
17681771 send := func(f ports.Frame) bool {
17691772 if err := transport.Send(f); err != nil {
@@ -1792,13 +1795,43 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
17921795 }
17931796 return send(ports.Frame{Type: ports.MsgAck, Payload: payload})
17941797 }
1798+ var heldInput *ports.Frame
17951799 for {
1800+ if heldInput != nil {
1801+ paused, changed := inputGate.snapshot()
1802+ if !paused {
1803+ if !send(*heldInput) {
1804+ return
1805+ }
1806+ heldInput = nil
1807+ continue
1808+ }
1809+ select {
1810+ case <-acks.wake:
1811+ if !sendAck() {
1812+ return
1813+ }
1814+ case frame := <-control:
1815+ if !send(frame) {
1816+ return
1817+ }
1818+ case <-changed:
1819+ case <-ctx.Done():
1820+ return
1821+ }
1822+ continue
1823+ }
17961824 select {
17971825 case <-acks.wake:
17981826 if !sendAck() {
17991827 return
18001828 }
18011829 continue
1830+ case frame := <-control:
1831+ if !send(frame) {
1832+ return
1833+ }
1834+ continue
18021835 default:
18031836 }
18041837 select {
@@ -1806,16 +1839,27 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
18061839 if !sendAck() {
18071840 return
18081841 }
1809- case f := <-in:
1842+ case frame := <-control:
1843+ if !send(frame) {
1844+ return
1845+ }
1846+ case frame := <-in:
18101847 select {
18111848 case <-ctx.Done():
18121849 return
18131850 default:
18141851 }
1815- if (f.Type == ports.MsgInput || f.Type == ports.MsgImagePush) && !inputGate.wait(ctx) {
1816- return
1852+ if frame.Type == ports.MsgInput || frame.Type == ports.MsgImagePush {
1853+ if paused, _ := inputGate.snapshot(); paused {
1854+ held := frame
1855+ heldInput = &held
1856+ if inputGate != nil && inputGate.afterInputHeld != nil {
1857+ inputGate.afterInputHeld()
1858+ }
1859+ continue
1860+ }
18171861 }
1818- if !send(f) {
1862+ if !send(frame) {
18191863 return
18201864 }
18211865 case <-ctx.Done():
@@ -1197,6 +1197,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1197 defer revokeInputClaim()1197 defer revokeInputClaim()
1198 1198
1199 sendCh := make(chan ports.Frame, sendQueueDepth)1199 sendCh := make(chan ports.Frame, sendQueueDepth)
1200+ controlCh := make(chan ports.Frame, 1)
1200 inputGate := newSamePeerInputGate()1201 inputGate := newSamePeerInputGate()
1201 ackQueue := newCumulativeAckQueue()1202 ackQueue := newCumulativeAckQueue()
1202 sendErrCh := make(chan error, 1)1203 sendErrCh := make(chan error, 1)
@@ -1338,7 +1339,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1338 // or unchanged screen may legitimately produce no initial Output frame.1339 // or unchanged screen may legitimately produce no initial Output frame.
1339 endHandshake()1340 endHandshake()
1340 senderStarted = true1341 senderStarted = true
1341- go runSender(loopCtx, cancel, transport, sendCh, inputGate, ackQueue, sendErrCh, log)1342+ go runSender(loopCtx, cancel, transport, controlCh, sendCh, inputGate, ackQueue, sendErrCh, log)
1342 stdinDone := make(chan struct{})1343 stdinDone := make(chan struct{})
1343 go func() {1344 go func() {
1344 defer close(stdinDone)1345 defer close(stdinDone)
@@ -1589,7 +1590,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1589 samePeerSwitch = &samePeerSwitchPending{requestID: nextSamePeerRequestID, target: *target.ExactTarget}1590 samePeerSwitch = &samePeerSwitchPending{requestID: nextSamePeerRequestID, target: *target.ExactTarget}
1590 inputGate.setPaused(true)1591 inputGate.setPaused(true)
1591 select {1592 select {
1592- case sendCh <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: payload}:1593+ case controlCh <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: payload}:
1593 continue1594 continue
1594 case <-loopCtx.Done():1595 case <-loopCtx.Done():
1595 return loopCanceledResult()1596 return loopCanceledResult()
@@ -1763,7 +1764,9 @@ func (q *cumulativeAckQueue) take() (uint64, uint64) {
1763 return epoch, state1764 return epoch, state
1764 }1765 }
1765 1766
1766-func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, inputGate *samePeerInputGate, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {1767+// runSender preserves normal-frame order while allowing switch control and
1768+// output ACKs to make progress when raw input is held for a same-peer switch.
1769+func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, control, in <-chan ports.Frame, inputGate *samePeerInputGate, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
1767 defer log.Debug("sender pump exited")1770 defer log.Debug("sender pump exited")
1768 send := func(f ports.Frame) bool {1771 send := func(f ports.Frame) bool {
1769 if err := transport.Send(f); err != nil {1772 if err := transport.Send(f); err != nil {
@@ -1792,13 +1795,43 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
1792 }1795 }
1793 return send(ports.Frame{Type: ports.MsgAck, Payload: payload})1796 return send(ports.Frame{Type: ports.MsgAck, Payload: payload})
1794 }1797 }
1798+ var heldInput *ports.Frame
1795 for {1799 for {
1800+ if heldInput != nil {
1801+ paused, changed := inputGate.snapshot()
1802+ if !paused {
1803+ if !send(*heldInput) {
1804+ return
1805+ }
1806+ heldInput = nil
1807+ continue
1808+ }
1809+ select {
1810+ case <-acks.wake:
1811+ if !sendAck() {
1812+ return
1813+ }
1814+ case frame := <-control:
1815+ if !send(frame) {
1816+ return
1817+ }
1818+ case <-changed:
1819+ case <-ctx.Done():
1820+ return
1821+ }
1822+ continue
1823+ }
1796 select {1824 select {
1797 case <-acks.wake:1825 case <-acks.wake:
1798 if !sendAck() {1826 if !sendAck() {
1799 return1827 return
1800 }1828 }
1801 continue1829 continue
1830+ case frame := <-control:
1831+ if !send(frame) {
1832+ return
1833+ }
1834+ continue
1802 default:1835 default:
1803 }1836 }
1804 select {1837 select {
@@ -1806,16 +1839,27 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
1806 if !sendAck() {1839 if !sendAck() {
1807 return1840 return
1808 }1841 }
1809- case f := <-in:1842+ case frame := <-control:
1843+ if !send(frame) {
1844+ return
1845+ }
1846+ case frame := <-in:
1810 select {1847 select {
1811 case <-ctx.Done():1848 case <-ctx.Done():
1812 return1849 return
1813 default:1850 default:
1814 }1851 }
1815- if (f.Type == ports.MsgInput || f.Type == ports.MsgImagePush) && !inputGate.wait(ctx) {1852+ if frame.Type == ports.MsgInput || frame.Type == ports.MsgImagePush {
1816- return1853+ if paused, _ := inputGate.snapshot(); paused {
1854+ held := frame
1855+ heldInput = &held
1856+ if inputGate != nil && inputGate.afterInputHeld != nil {
1857+ inputGate.afterInputHeld()
1858+ }
1859+ continue
1860+ }
1817 }1861 }
1818- if !send(f) {1862+ if !send(frame) {
1819 return1863 return
1820 }1864 }
1821 case <-ctx.Done():1865 case <-ctx.Done():
M internal/usecase/client/same_peer_switch.go +11 -5
@@ -14,6 +14,8 @@ type samePeerInputGate struct {
1414 mu sync.Mutex
1515 paused bool
1616 changed chan struct{}
17+ // afterInputHeld is a deterministic test synchronization seam.
18+ afterInputHeld func()
1719 }
1820
1921 type samePeerSwitchPending struct {
@@ -38,14 +40,18 @@ func (g *samePeerInputGate) setPaused(paused bool) {
3840 g.mu.Unlock()
3941 }
4042
41-func (g *samePeerInputGate) wait(ctx context.Context) bool {
43+func (g *samePeerInputGate) snapshot() (bool, <-chan struct{}) {
4244 if g == nil {
43- return true
45+ return false, nil
4446 }
47+ g.mu.Lock()
48+ defer g.mu.Unlock()
49+ return g.paused, g.changed
50+}
51+
52+func (g *samePeerInputGate) wait(ctx context.Context) bool {
4553 for {
46- g.mu.Lock()
47- paused, changed := g.paused, g.changed
48- g.mu.Unlock()
54+ paused, changed := g.snapshot()
4955 if !paused {
5056 return true
5157 }
@@ -14,6 +14,8 @@ type samePeerInputGate struct {
14 mu sync.Mutex14 mu sync.Mutex
15 paused bool15 paused bool
16 changed chan struct{}16 changed chan struct{}
17+ // afterInputHeld is a deterministic test synchronization seam.
18+ afterInputHeld func()
17 }19 }
18 20
19 type samePeerSwitchPending struct {21 type samePeerSwitchPending struct {
@@ -38,14 +40,18 @@ func (g *samePeerInputGate) setPaused(paused bool) {
38 g.mu.Unlock()40 g.mu.Unlock()
39 }41 }
40 42
41-func (g *samePeerInputGate) wait(ctx context.Context) bool {43+func (g *samePeerInputGate) snapshot() (bool, <-chan struct{}) {
42 if g == nil {44 if g == nil {
43- return true45+ return false, nil
44 }46 }
47+ g.mu.Lock()
48+ defer g.mu.Unlock()
49+ return g.paused, g.changed
50+}
51+
52+func (g *samePeerInputGate) wait(ctx context.Context) bool {
45 for {53 for {
46- g.mu.Lock()54+ paused, changed := g.snapshot()
47- paused, changed := g.paused, g.changed
48- g.mu.Unlock()
49 if !paused {55 if !paused {
50 return true56 return true
51 }57 }
M internal/usecase/daemon/client_transport_errors.go +1 -0
@@ -170,6 +170,7 @@ func (d *Daemon) reserveAttachmentSendErrorCleanup(token attachmentConnectionTok
170170 }
171171
172172 func (d *Daemon) finishSendErrorDetach(sess *session, ac *attachedClient, failed ports.Transport) {
173+ ac.clearSamePeerOffer()
173174 name := sess.nameSnapshot()
174175 if rc := sess.renderCoordinator(); rc != nil {
175176 rc.noteDetach(ac)
@@ -170,6 +170,7 @@ func (d *Daemon) reserveAttachmentSendErrorCleanup(token attachmentConnectionTok
170 }170 }
171 171
172 func (d *Daemon) finishSendErrorDetach(sess *session, ac *attachedClient, failed ports.Transport) {172 func (d *Daemon) finishSendErrorDetach(sess *session, ac *attachedClient, failed ports.Transport) {
173+ ac.clearSamePeerOffer()
173 name := sess.nameSnapshot()174 name := sess.nameSnapshot()
174 if rc := sess.renderCoordinator(); rc != nil {175 if rc := sess.renderCoordinator(); rc != nil {
175 rc.noteDetach(ac)176 rc.noteDetach(ac)
M internal/usecase/daemon/same_peer_switch_test.go +12 -0
@@ -53,6 +53,18 @@ func TestSamePeerSwitchTransitionsExactTargetAndPreferredTab(t *testing.T) {
5353 require.Equal(t, ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "target"}, identity.Target)
5454 }
5555
56+func TestFinishSendErrorDetachClearsSamePeerOffer(t *testing.T) {
57+ d, sess, ac, _, releases := newManualTabSession(t, 1)
58+ defer releaseAll(releases)
59+ ac.offerSamePeerTarget(ports.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "target"})
60+
61+ d.finishSendErrorDetach(sess, ac, ac.transport())
62+
63+ ac.samePeerOfferMu.Lock()
64+ defer ac.samePeerOfferMu.Unlock()
65+ require.Nil(t, ac.samePeerOffer)
66+}
67+
5668 func TestSamePeerSwitchRejectsStaleTargetWithoutMutation(t *testing.T) {
5769 d, source, ac, sends, releases := newManualTabSession(t, 1)
5870 defer releaseAll(releases)
@@ -53,6 +53,18 @@ func TestSamePeerSwitchTransitionsExactTargetAndPreferredTab(t *testing.T) {
53 require.Equal(t, ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "target"}, identity.Target)53 require.Equal(t, ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "target"}, identity.Target)
54 }54 }
55 55
56+func TestFinishSendErrorDetachClearsSamePeerOffer(t *testing.T) {
57+ d, sess, ac, _, releases := newManualTabSession(t, 1)
58+ defer releaseAll(releases)
59+ ac.offerSamePeerTarget(ports.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "target"})
60+
61+ d.finishSendErrorDetach(sess, ac, ac.transport())
62+
63+ ac.samePeerOfferMu.Lock()
64+ defer ac.samePeerOfferMu.Unlock()
65+ require.Nil(t, ac.samePeerOffer)
66+}
67+
56 func TestSamePeerSwitchRejectsStaleTargetWithoutMutation(t *testing.T) {68 func TestSamePeerSwitchRejectsStaleTargetWithoutMutation(t *testing.T) {
57 d, source, ac, sends, releases := newManualTabSession(t, 1)69 d, source, ac, sends, releases := newManualTabSession(t, 1)
58 defer releaseAll(releases)70 defer releaseAll(releases)

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close