ack_queue_test.go RawHistoryFolder

1package client
2
3import (
4	"context"
5	"io"
6	"log/slog"
7	"testing"
8
9	"github.com/bnema/vev/internal/ports"
10	"github.com/stretchr/testify/require"
11)
12
13type ackRecordingTransport struct{ sent chan ports.Frame }
14
15func (t *ackRecordingTransport) Send(f ports.Frame) error   { t.sent <- f; return nil }
16func (t *ackRecordingTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
17func (t *ackRecordingTransport) Close() error               { return nil }
18
19func TestCumulativeAckBypassesFullNormalSendQueue(t *testing.T) {
20	normal := make(chan ports.Frame, sendQueueDepth)
21	for range cap(normal) {
22		normal <- ports.Frame{Type: ports.MsgInput}
23	}
24	const epoch = 7
25	acks := newCumulativeAckQueue()
26	for state := uint64(1); state <= maxUnackedOutputStatesForTest; state++ {
27		acks.offer(epoch, state)
28	}
29
30	ctx, cancel := context.WithCancel(context.Background())
31	defer cancel()
32	tr := &ackRecordingTransport{sent: make(chan ports.Frame, sendQueueDepth+1)}
33	errCh := make(chan error, 1)
34	go runSender(ctx, cancel, tr, normal, acks, errCh, slog.Default())
35
36	first := <-tr.sent
37	require.Equal(t, ports.MsgAck, first.Type)
38	ack, err := ports.UnmarshalAck(first.Payload)
39	require.NoError(t, err)
40	require.Equal(t, uint64(epoch), ack.Epoch)
41	require.Equal(t, uint64(maxUnackedOutputStatesForTest), ack.State)
42}
43
44const maxUnackedOutputStatesForTest = 8