Commilitia-Drop/internal/relay/ring_test.go

package relay

import (
	"bytes"
	"context"
	"errors"
	"io"
	"sync"
	"testing"
	"time"
)

func TestRing_BasicWriteRead(t *testing.T) {
	r := newRing(64)
	want := []byte("hello, world")
	if n, err := r.Write(context.Background(), want); err != nil || n != len(want) {
		t.Fatalf("write: n=%d err=%v", n, err)
	}
	got := make([]byte, 32)
	n, err := r.Read(context.Background(), got)
	if err != nil {
		t.Fatalf("read: %v", err)
	}
	if !bytes.Equal(got[:n], want) {
		t.Errorf("read mismatch: got %q want %q", got[:n], want)
	}
}

func TestRing_CloseWriterSignalsEOF(t *testing.T) {
	r := newRing(32)
	_, _ = r.Write(context.Background(), []byte("abc"))
	r.CloseWriter()

	got := make([]byte, 16)
	n, err := r.Read(context.Background(), got)
	if err != nil || n != 3 {
		t.Fatalf("first read: n=%d err=%v", n, err)
	}
	n, err = r.Read(context.Background(), got)
	if !errors.Is(err, io.EOF) {
		t.Errorf("expected EOF after drain; got n=%d err=%v", n, err)
	}
}

func TestRing_AbortReturnsErrClosed(t *testing.T) {
	r := newRing(32)
	r.Abort()
	if _, err := r.Write(context.Background(), []byte("x")); !errors.Is(err, ErrClosed) {
		t.Errorf("write after abort: got %v want ErrClosed", err)
	}
	if _, err := r.Read(context.Background(), make([]byte, 1)); !errors.Is(err, ErrClosed) {
		t.Errorf("read after abort: got %v want ErrClosed", err)
	}
}

func TestRing_BlocksWhenFullThenUnblocksOnRead(t *testing.T) {
	r := newRing(8)
	if _, err := r.Write(context.Background(), []byte("12345678")); err != nil {
		t.Fatalf("fill: %v", err)
	}

	done := make(chan struct{})
	go func() {
		_, err := r.Write(context.Background(), []byte("X"))
		if err != nil {
			t.Errorf("blocked write: %v", err)
		}
		close(done)
	}()

	// Without a reader the goroutine should block.
	select {
	case <-done:
		t.Fatal("write should block when ring is full")
	case <-time.After(50 * time.Millisecond):
	}

	got := make([]byte, 4)
	if _, err := r.Read(context.Background(), got); err != nil {
		t.Fatalf("read: %v", err)
	}
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("write did not unblock after read")
	}
}

func TestRing_CtxCancelsBlockedRead(t *testing.T) {
	r := newRing(8)
	ctx, cancel := context.WithCancel(context.Background())

	type res struct {
		n   int
		err error
	}
	out := make(chan res, 1)
	go func() {
		buf := make([]byte, 4)
		n, err := r.Read(ctx, buf)
		out <- res{n, err}
	}()

	time.Sleep(20 * time.Millisecond)
	cancel()

	select {
	case r := <-out:
		if !errors.Is(r.err, context.Canceled) {
			t.Errorf("expected context.Canceled, got %v", r.err)
		}
	case <-time.After(time.Second):
		t.Fatal("blocked read did not unblock on ctx cancel")
	}
}

func TestRing_WrapsAround(t *testing.T) {
	r := newRing(4)
	// Cycle through enough bytes to force head/tail wrapping.
	for i := 0; i < 20; i++ {
		_, _ = r.Write(context.Background(), []byte{byte(i)})
		got := make([]byte, 1)
		n, _ := r.Read(context.Background(), got)
		if n != 1 || got[0] != byte(i) {
			t.Errorf("iter %d: got %v want [%d]", i, got, i)
		}
	}
}

func TestRing_StressManyChunks(t *testing.T) {
	const total = 4 * 1024
	r := newRing(64)

	var wgWriter, wgReader sync.WaitGroup
	wgWriter.Add(1)
	wgReader.Add(1)

	src := make([]byte, total)
	for i := range src {
		src[i] = byte(i)
	}

	go func() {
		defer wgWriter.Done()
		for i := 0; i < total; i += 13 {
			end := i + 13
			if end > total {
				end = total
			}
			_, _ = r.Write(context.Background(), src[i:end])
		}
		r.CloseWriter()
	}()

	got := make([]byte, 0, total)
	go func() {
		defer wgReader.Done()
		buf := make([]byte, 17)
		for {
			n, err := r.Read(context.Background(), buf)
			got = append(got, buf[:n]...)
			if errors.Is(err, io.EOF) {
				return
			}
			if err != nil {
				t.Errorf("read: %v", err)
				return
			}
		}
	}()

	wgWriter.Wait()
	wgReader.Wait()
	if !bytes.Equal(got, src) {
		t.Errorf("mismatch: got %d bytes, want %d", len(got), len(src))
	}
}