rendezvous_test.go

package rchan

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/redis/go-redis/v9"
)

func TestRendezvousHandoff(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	ch, err := New[string](ctx, rdb, testName(t, rdb))
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	done := make(chan error, 1)
	go func() { done <- ch.Send(ctx, "v") }()
	select {
	case err := <-done:
		t.Fatalf("rendezvous send must block until pickup, returned %v", err)
	case <-time.After(200 * time.Millisecond):
	}
	rctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	v, ok, err := ch.Recv(rctx)
	if err != nil || !ok || v != "v" {
		t.Fatalf("v=%q ok=%v err=%v", v, ok, err)
	}
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("sender did not complete cleanly: %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("sender still blocked after pickup")
	}
}

func TestRendezvousCancelRetracts(t *testing.T) {
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, _ := New[string](context.Background(), rdb, name)
	ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
	defer cancel()
	if err := ch.Send(ctx, "v"); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("want DeadlineExceeded, got %v", err)
	}
	n, _ := rdb.XLen(context.Background(), "rchan:{"+name+"}:s").Result()
	if n != 0 {
		t.Fatalf("XLEN=%d, want 0 (entry retracted)", n)
	}
}

func TestRetractLosesToDelivery(t *testing.T) {
	// Script-level check of the race branch: once an entry is in the PEL,
	// retract must refuse so the sender treats the value as delivered.
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, _ := New[string](ctx, rdb, name)
	_ = ch
	stream := "rchan:{" + name + "}:s"
	id, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream: stream, Values: map[string]any{"v": `"x"`, "rz": "r1"},
	}).Result()
	if err != nil {
		t.Fatal(err)
	}
	// Deliver it (into the PEL) with a raw consumer.
	if err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group: "recv", Consumer: "rogue", Streams: []string{stream, ">"},
		Count: 1, Block: -1,
	}).Err(); err != nil {
		t.Fatal(err)
	}
	got, err := retractScript.Run(ctx, rdb, []string{stream}, "recv", id).Bool()
	if err != nil {
		t.Fatal(err)
	}
	if got {
		t.Fatal("retract must fail for a delivered (PEL) entry")
	}
}

func TestRendezvousSurvivesReceiverCrash(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	vis := 200 * time.Millisecond
	ch, _ := New[string](ctx, rdb, name, WithVisibilityTimeout(vis))
	defer ch.Stop()
	stream := "rchan:{" + name + "}:s"

	done := make(chan error, 1)
	go func() { done <- ch.Send(ctx, "v") }()
	// Rogue consumer picks the entry up and "crashes" before replying.
	eventually(t, 2*time.Second, "entry parked", func() bool {
		n, _ := rdb.XLen(ctx, stream).Result()
		return n == 1
	})
	if err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group: "recv", Consumer: "rogue", Streams: []string{stream, ">"},
		Count: 1, Block: -1,
	}).Err(); err != nil {
		t.Fatal(err)
	}
	// A healthy receiving handle reclaims it and completes the rendezvous.
	rch, _ := New[string](ctx, rdb, name, WithVisibilityTimeout(vis))
	defer rch.Stop()
	rctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	v, ok, err := rch.Recv(rctx)
	if err != nil || !ok || v != "v" {
		t.Fatalf("v=%q ok=%v err=%v", v, ok, err)
	}
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("sender should complete after reclaim, got %v", err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("sender still blocked; reclaim did not send rendezvous reply")
	}
}

func TestRendezvousSenderUnblocksOnDestroy(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, err := New[string](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	done := make(chan error, 1)
	go func() { done <- ch.Send(ctx, "doomed") }()
	eventually(t, 2*time.Second, "entry parked", func() bool {
		n, _ := rdb.XLen(ctx, "rchan:{"+name+"}:s").Result()
		return n == 1
	})
	if err := Destroy(ctx, rdb, name); err != nil {
		t.Fatal(err)
	}
	select {
	case err := <-done:
		if !errors.Is(err, ErrClosed) {
			t.Fatalf("want ErrClosed after destroy, got %v", err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("rendezvous sender still parked after Destroy")
	}
}

func TestRendezvousWaitsForLateReceiver(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	ch, err := New[string](ctx, rdb, testName(t, rdb))
	if err != nil {
		t.Fatal(err)
	}
	defer ch.Stop()
	done := make(chan error, 1)
	go func() { done <- ch.Send(ctx, "patience") }()
	time.Sleep(2500 * time.Millisecond) // longer than two BLPop slices
	select {
	case err := <-done:
		t.Fatalf("send gave up while no receiver present: %v", err)
	default:
	}
	rctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	v, ok, err := ch.Recv(rctx)
	if err != nil || !ok || v != "patience" {
		t.Fatalf("v=%q ok=%v err=%v", v, ok, err)
	}
	if err := <-done; err != nil {
		t.Fatalf("sender error after late pickup: %v", err)
	}
}

// TestRendezvousParkedSenderDrainsAfterClose: Close does not wake a parked
// rendezvous sender — its value stays drainable, a receiver completes the
// handoff, and only then is the channel observed as closed.
func TestRendezvousParkedSenderDrainsAfterClose(t *testing.T) {
	ctx := context.Background()
	rdb := testClient(t)
	name := testName(t, rdb)
	ch, err := New[string](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	done := make(chan error, 1)
	go func() { done <- ch.Send(ctx, "parked") }()
	eventually(t, 2*time.Second, "entry parked", func() bool {
		n, _ := rdb.XLen(ctx, "rchan:{"+name+"}:s").Result()
		return n == 1
	})
	if err := ch.Close(ctx); err != nil {
		t.Fatal(err)
	}
	select {
	case err := <-done:
		t.Fatalf("close must not wake a parked rendezvous sender, got %v", err)
	case <-time.After(1500 * time.Millisecond):
	}
	rch, err := New[string](ctx, rdb, name)
	if err != nil {
		t.Fatal(err)
	}
	defer rch.Stop()
	rctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	v, ok, err := rch.Recv(rctx)
	if err != nil || !ok || v != "parked" {
		t.Fatalf("v=%q ok=%v err=%v", v, ok, err)
	}
	if err := <-done; err != nil {
		t.Fatalf("sender should complete after drain, got %v", err)
	}
	if _, ok, err := rch.Recv(rctx); ok || err != nil {
		t.Fatalf("after drain: ok=%v err=%v, want closed", ok, err)
	}
}