send.go

package rchan

import (
	"context"
	"errors"
	"fmt"
	"time"

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

// Send delivers v to exactly one receiver. It blocks while the buffer is
// full (capacity > 0) or until pickup (capacity 0, rendezvous).
func (c *Chan[T]) Send(ctx context.Context, v T) error {
	data, err := c.opts.codec.Marshal(v)
	if err != nil {
		return fmt.Errorf("rchan: encode: %w", err)
	}
	if c.opts.capacity == 0 {
		return c.sendRendezvous(ctx, data)
	}
	return c.sendBuffered(ctx, data)
}

func (c *Chan[T]) trySend(ctx context.Context, data []byte, rz string) (string, error) {
	return sendScript.Run(ctx, c.rdb,
		[]string{c.keys.meta, c.keys.stream},
		c.opts.capacity, data, rz).Text()
}

func (c *Chan[T]) sendBuffered(ctx context.Context, data []byte) error {
	res, err := c.trySend(ctx, data, "")
	if err != nil {
		return fmt.Errorf("rchan: send: %w", err)
	}
	switch res {
	case "CLOSED":
		return ErrClosed
	case "FULL":
		return c.sendBufferedWait(ctx, data)
	default:
		return nil
	}
}

// sendBufferedWait retries the gate until a slot frees. It subscribes to the
// sig channel before re-attempting so a 'space' publish between attempts is
// not missed; a jittered poll is the safety net.
func (c *Chan[T]) sendBufferedWait(ctx context.Context, data []byte) error {
	sub := c.rdb.Subscribe(ctx, c.keys.sig)
	defer sub.Close()
	signals := sub.Channel()
	for {
		res, err := c.trySend(ctx, data, "")
		if err != nil {
			return fmt.Errorf("rchan: send: %w", err)
		}
		switch res {
		case "CLOSED":
			return ErrClosed
		case "FULL":
		default:
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-signals: // 'space' or 'close' — either way, retry the gate
		case <-time.After(jitter(c.opts.poll)):
		}
	}
}

// sendRendezvous parks the value with a reply id and blocks until a receiver
// picks it up (LPUSH on the reply key), like an unbuffered chan send.
func (c *Chan[T]) sendRendezvous(ctx context.Context, data []byte) error {
	rz := randHex(16)
	res, err := c.trySend(ctx, data, rz)
	if err != nil {
		return fmt.Errorf("rchan: send: %w", err)
	}
	if res == "CLOSED" {
		return ErrClosed
	}
	entryID := res
	rzk := rzKey(c.name, rz)
	var waitErr error
	for waitErr == nil {
		// Short BLPop slices: go-redis blocking commands do not abort on ctx
		// cancellation mid-read (ContextTimeoutEnabled is off by default),
		// so block 1s at a time and check ctx between slices.
		_, err = c.rdb.BLPop(ctx, time.Second, rzk).Result()
		if err == nil {
			return nil // reply received: rendezvous complete
		}
		switch {
		case !errors.Is(err, redis.Nil) && ctx.Err() == nil:
			waitErr = fmt.Errorf("rchan: rendezvous wait: %w", err)
		case ctx.Err() != nil:
			waitErr = ctx.Err()
		default:
			// Still parked: make sure the channel still exists. Destroy
			// unlinks every key, and nothing would ever push our reply.
			if n, eerr := c.rdb.Exists(ctx, c.keys.meta).Result(); eerr == nil && n == 0 {
				waitErr = ErrClosed
			}
		}
	}
	// Exiting without a reply (cancel, transient error, or destroy): retract,
	// unless a receiver already has it — never leave a parked entry behind.
	bg, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer cancel()
	retracted, rerr := retractScript.Run(bg, c.rdb,
		[]string{c.keys.stream}, group, entryID).Bool()
	if rerr != nil {
		return errors.Join(waitErr, fmt.Errorf("rchan: retract: %w", rerr))
	}
	if retracted {
		return waitErr
	}
	if errors.Is(waitErr, ErrClosed) {
		// Destroyed: the entry is gone because the stream was unlinked, not
		// because a receiver took it — the reply will never arrive.
		return ErrClosed
	}
	// Delivered while we were giving up: the send happened. Drop the reply
	// key; a straggling LPUSH dies via its TTL.
	c.rdb.Unlink(bg, rzk)
	return nil
}