diff --git a/pkg/moq/moq_test.go b/pkg/moq/moq_test.go index 351843ddf..ef1ec019b 100644 --- a/pkg/moq/moq_test.go +++ b/pkg/moq/moq_test.go @@ -264,9 +264,10 @@ func newFakeGroup(seq uint64, payload string) groupStream { // Groups arrive on independent streams in whatever order the network and // the transport's accept queue produce, and a subscription hands them out -// in sequence order regardless: a straggler is waited for, a group the -// publisher dropped is not, and one that never shows up is given up on -// after gapTimeout rather than stalling the track forever. +// in sequence order regardless: groups that outrun SUBSCRIBE_OK wait for +// the start it names, a straggler is waited for, a group the publisher +// dropped is not, and one that never shows up is given up on after +// gapTimeout rather than stalling the track forever. func TestSubscriptionOrdersGroups(t *testing.T) { gapTimeout = 100 * time.Millisecond ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) @@ -274,7 +275,7 @@ func TestSubscriptionOrdersGroups(t *testing.T) { sctx, scancel := context.WithCancelCause(ctx) defer scancel(nil) s := &Session{ctx: sctx, cancel: scancel, subs: map[uint64]*Subscription{}} - sub := &Subscription{s: s, done: make(chan struct{}), frames: make(chan *Frame, 16), notify: make(chan struct{}, 1), started: true} + sub := &Subscription{s: s, done: make(chan struct{}), frames: make(chan *Frame, 16), notify: make(chan struct{}, 1)} go sub.run() defer sub.Close() @@ -285,14 +286,24 @@ func TestSubscriptionOrdersGroups(t *testing.T) { return f.Group } - // Out of order: 2 and 1 wait for 0. + // Out of order, before SUBSCRIBE_OK: 2 and 1 wait for the answer, then + // for the 0 it names. sub.enqueue(newFakeGroup(2, "g2")) sub.enqueue(newFakeGroup(1, "g1")) - select { - case f := <-sub.frames: - t.Fatalf("group %d delivered before group 0", f.Group) - case <-time.After(50 * time.Millisecond): + noFrame := func() { + t.Helper() + select { + case f := <-sub.frames: + t.Fatalf("group %d delivered before group 0", f.Group) + case <-time.After(20 * time.Millisecond): // both well inside gapTimeout + } } + noFrame() + sub.mu.Lock() + sub.next, sub.started = 0, true // SUBSCRIBE_OK naming group 0 + sub.mu.Unlock() + sub.signal() + noFrame() sub.enqueue(newFakeGroup(0, "g0")) require.Equal(t, uint64(0), next()) require.Equal(t, uint64(1), next()) diff --git a/pkg/moq/subscribe.go b/pkg/moq/subscribe.go index e316d3958..9302b4172 100644 --- a/pkg/moq/subscribe.go +++ b/pkg/moq/subscribe.go @@ -94,9 +94,9 @@ type Subscription struct { mu sync.Mutex // queue holds accepted group streams by ascending sequence; next is - // the sequence delivered next (once started, by SUBSCRIBE_OK or the - // first group), so queue[0].seq > next is a hole. active is the - // stream being read, so Close can cut a read short. + // the sequence delivered next (once started, by SUBSCRIBE_OK), so + // queue[0].seq > next is a hole. active is the stream being read, so + // Close can cut a read short. queue []groupStream active recvStream next uint64 @@ -279,10 +279,18 @@ const ( func (sub *Subscription) pick() (groupStream, pickState) { sub.mu.Lock() defer sub.mu.Unlock() - // SUBSCRIBE_OK and the first group travel on different streams; if - // the group wins the race it resolves the start just as well. - if !sub.started && len(sub.queue) > 0 { - sub.next, sub.started = sub.queue[0].seq, true + // SUBSCRIBE_OK names the start group, but groups travel on their own + // streams and can arrive first: starting at whichever won the race + // would skip the groups before it. Wait for the answer; a publisher + // that never sends one gets the gap timeout (see skipGap). + if !sub.started { + if len(sub.queue) > 0 { + return groupStream{}, pickGap + } + if sub.ended { + return groupStream{}, pickOver + } + return groupStream{}, pickWait } // Groups the publisher dropped are not waited for; ranges may chain. for advanced := true; advanced; { @@ -316,14 +324,15 @@ func (sub *Subscription) pick() (groupStream, pickState) { return groupStream{}, pickWait } -// skipGap gives up on the group(s) delivery is waiting for: the next -// queued group becomes next, or the track ends if nothing is queued. +// skipGap gives up on the group(s) delivery is waiting for, SUBSCRIBE_OK +// included: the next queued group becomes next, or the track ends if +// nothing is queued. func (sub *Subscription) skipGap() { sub.mu.Lock() defer sub.mu.Unlock() if len(sub.queue) > 0 { log.Warn(sub.s.ctx, "moq: groups never arrived, skipping", "from", sub.next, "to", sub.queue[0].seq-1) - sub.next = sub.queue[0].seq + sub.next, sub.started = sub.queue[0].seq, true return } if sub.ended { @@ -382,7 +391,13 @@ func (sub *Subscription) watchControl(br *bufio.Reader) { if err != nil { sub.mu.Lock() if !errors.Is(err, io.EOF) && sub.err == nil { - sub.err = fmt.Errorf("subscription reset: %w", err) + // The session ending also ends this stream; its cause, not + // the stream error it leaves, is why the subscription is over. + if serr := sub.s.Err(); serr != nil { + sub.err = serr + } else { + sub.err = fmt.Errorf("subscription reset: %w", err) + } } sub.ended = true sub.mu.Unlock()