detached 00fabb2

test(brokeripc): cover credit grant dispatch and park waiters before close

commit
00fabb21997fe26f5399d518f4d1c88f1aa30137
author
brice · · 2026-10-08 20:07 UTC
parent
16716df
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +82 -0
Unified
UnifiedSplit
A internal/adapters/brokeripc/credit_session_test.go +72 -0
@@ -0,0 +1,72 @@
1+package brokeripc
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/stretchr/testify/require"
8+
9+ "github.com/bnema/vev/internal/adapters/brokerwire"
10+ "github.com/bnema/vev/internal/ports"
11+)
12+
13+// TestStreamCreditGrantDispatch proves how the session settles one client
14+// credit grant: a grant past the window settles only its stream, a grant for a
15+// retired stream is discarded, and a grant for a stream that was never opened
16+// fails the whole connection.
17+func TestStreamCreditGrantDispatch(t *testing.T) {
18+ open := func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope, id ports.BrokerStreamID) {
19+ t.Helper()
20+ raw.send(t, brokerwire.OpenStream{
21+ Epoch: scope.Epoch, Connection: scope.Connection, Stream: id,
22+ Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(),
23+ StartMode: ports.BrokerDaemonStartIfNeeded,
24+ })
25+ opened, ok := raw.recv(t).(brokerwire.StreamOpened)
26+ require.True(t, ok, "stream %d must open", id)
27+ require.Equal(t, id, opened.Stream)
28+ }
29+ grant := func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope, id ports.BrokerStreamID) {
30+ t.Helper()
31+ raw.send(t, brokerwire.StreamWindowUpdate{Epoch: scope.Epoch, Connection: scope.Connection, Stream: id, Credit: 1})
32+ }
33+ requireStreamFailed := func(t *testing.T, raw *rawCarriage, id ports.BrokerStreamID) {
34+ t.Helper()
35+ closed, ok := raw.recv(t).(brokerwire.StreamClosed)
36+ require.True(t, ok, "an over-window grant must be answered with StreamClosed")
37+ require.Equal(t, id, closed.Stream)
38+ require.True(t, closed.HasError)
39+ }
40+
41+ for _, tc := range []struct {
42+ name string
43+ run func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope)
44+ }{
45+ {"grant over the window settles only its stream", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
46+ open(t, raw, scope, 1)
47+ open(t, raw, scope, 2)
48+ // A fresh stream already holds the whole window, so any grant overflows.
49+ grant(t, raw, scope, 1)
50+ requireStreamFailed(t, raw, 1)
51+ open(t, raw, scope, 3) // the connection still serves new streams
52+ }},
53+ {"grant for a retired stream is discarded", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
54+ open(t, raw, scope, 1)
55+ grant(t, raw, scope, 1)
56+ requireStreamFailed(t, raw, 1)
57+ grant(t, raw, scope, 1) // stream 1 is retired now
58+ open(t, raw, scope, 2)
59+ }},
60+ {"grant for a never-opened stream fails the connection", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
61+ open(t, raw, scope, 1)
62+ grant(t, raw, scope, 9)
63+ require.Error(t, raw.awaitReadError(t, 5*time.Second))
64+ }},
65+ } {
66+ t.Run(tc.name, func(t *testing.T) {
67+ e := startEndpoint(t, Config{})
68+ raw := rawDial(t, e, brokerwire.DefaultCeilings())
69+ tc.run(t, raw, raw.register(t))
70+ })
71+ }
72+}
@@ -0,0 +1,72 @@
1+package brokeripc
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/stretchr/testify/require"
8+
9+ "github.com/bnema/vev/internal/adapters/brokerwire"
10+ "github.com/bnema/vev/internal/ports"
11+)
12+
13+// TestStreamCreditGrantDispatch proves how the session settles one client
14+// credit grant: a grant past the window settles only its stream, a grant for a
15+// retired stream is discarded, and a grant for a stream that was never opened
16+// fails the whole connection.
17+func TestStreamCreditGrantDispatch(t *testing.T) {
18+ open := func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope, id ports.BrokerStreamID) {
19+ t.Helper()
20+ raw.send(t, brokerwire.OpenStream{
21+ Epoch: scope.Epoch, Connection: scope.Connection, Stream: id,
22+ Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(),
23+ StartMode: ports.BrokerDaemonStartIfNeeded,
24+ })
25+ opened, ok := raw.recv(t).(brokerwire.StreamOpened)
26+ require.True(t, ok, "stream %d must open", id)
27+ require.Equal(t, id, opened.Stream)
28+ }
29+ grant := func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope, id ports.BrokerStreamID) {
30+ t.Helper()
31+ raw.send(t, brokerwire.StreamWindowUpdate{Epoch: scope.Epoch, Connection: scope.Connection, Stream: id, Credit: 1})
32+ }
33+ requireStreamFailed := func(t *testing.T, raw *rawCarriage, id ports.BrokerStreamID) {
34+ t.Helper()
35+ closed, ok := raw.recv(t).(brokerwire.StreamClosed)
36+ require.True(t, ok, "an over-window grant must be answered with StreamClosed")
37+ require.Equal(t, id, closed.Stream)
38+ require.True(t, closed.HasError)
39+ }
40+
41+ for _, tc := range []struct {
42+ name string
43+ run func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope)
44+ }{
45+ {"grant over the window settles only its stream", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
46+ open(t, raw, scope, 1)
47+ open(t, raw, scope, 2)
48+ // A fresh stream already holds the whole window, so any grant overflows.
49+ grant(t, raw, scope, 1)
50+ requireStreamFailed(t, raw, 1)
51+ open(t, raw, scope, 3) // the connection still serves new streams
52+ }},
53+ {"grant for a retired stream is discarded", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
54+ open(t, raw, scope, 1)
55+ grant(t, raw, scope, 1)
56+ requireStreamFailed(t, raw, 1)
57+ grant(t, raw, scope, 1) // stream 1 is retired now
58+ open(t, raw, scope, 2)
59+ }},
60+ {"grant for a never-opened stream fails the connection", func(t *testing.T, raw *rawCarriage, scope brokerwire.Scope) {
61+ open(t, raw, scope, 1)
62+ grant(t, raw, scope, 9)
63+ require.Error(t, raw.awaitReadError(t, 5*time.Second))
64+ }},
65+ } {
66+ t.Run(tc.name, func(t *testing.T) {
67+ e := startEndpoint(t, Config{})
68+ raw := rawDial(t, e, brokerwire.DefaultCeilings())
69+ tc.run(t, raw, raw.register(t))
70+ })
71+ }
72+}
M internal/adapters/brokeripc/limits_test.go +5 -0
@@ -163,6 +163,11 @@ func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
163163 _, err := parked.Read(make([]byte, 3))
164164 done <- err
165165 }()
166+ select {
167+ case err := <-done:
168+ t.Fatalf("Read returned %v on an empty pipe", err)
169+ case <-time.After(20 * time.Millisecond):
170+ }
166171 parked.closeWith(ErrStreamCredit)
167172 select {
168173 case err := <-done:
@@ -163,6 +163,11 @@ func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
163 _, err := parked.Read(make([]byte, 3))163 _, err := parked.Read(make([]byte, 3))
164 done <- err164 done <- err
165 }()165 }()
166+ select {
167+ case err := <-done:
168+ t.Fatalf("Read returned %v on an empty pipe", err)
169+ case <-time.After(20 * time.Millisecond):
170+ }
166 parked.closeWith(ErrStreamCredit)171 parked.closeWith(ErrStreamCredit)
167 select {172 select {
168 case err := <-done:173 case err := <-done:
M internal/adapters/brokeripc/pipe_test.go +5 -0
@@ -194,6 +194,11 @@ func TestStreamPipeCredit(t *testing.T) {
194194 }
195195 reserved := make(chan error, 1)
196196 go func() { reserved <- p.reserve(chunk) }()
197+ select {
198+ case err := <-reserved:
199+ t.Fatalf("reserve returned %v without credit", err)
200+ case <-time.After(20 * time.Millisecond):
201+ }
197202 p.closeWith(ErrStreamGone)
198203 select {
199204 case err := <-reserved:
@@ -194,6 +194,11 @@ func TestStreamPipeCredit(t *testing.T) {
194 }194 }
195 reserved := make(chan error, 1)195 reserved := make(chan error, 1)
196 go func() { reserved <- p.reserve(chunk) }()196 go func() { reserved <- p.reserve(chunk) }()
197+ select {
198+ case err := <-reserved:
199+ t.Fatalf("reserve returned %v without credit", err)
200+ case <-time.After(20 * time.Millisecond):
201+ }
197 p.closeWith(ErrStreamGone)202 p.closeWith(ErrStreamGone)
198 select {203 select {
199 case err := <-reserved:204 case err := <-reserved:

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