func TasksWithID(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	queue := path.Join(qPrefix, "tasks_with_id")
	ids := []uuid.UUID{
		uuid.New(),
		uuid.New(),
		uuid.New(),
		uuid.New(),
	}
	var args []entroq.ModifyArg
	for _, id := range ids {
		args = append(args, entroq.InsertingInto(queue, entroq.WithID(id)))
	}
	inserted, _, err := client.Modify(ctx, args...)
	if err != nil {
		t.Fatalf("Insertion failed: %v", err)
	}
	if want, got := len(ids), len(inserted); want != got {
		t.Fatalf("Expected %d tasks inserted, got %d", want, got)
	}
	for i, task := range inserted {
		if want, got := ids[i], task.ID; want != got {
			t.Fatalf("Inserted task should have ID %q, but has %q", want, got)
		}
	}
	tasks, err := client.Tasks(ctx, queue)
	if err != nil {
		t.Fatalf("Error getting tasks from queue %q: %v", queue, err)
	}
	if want, got := len(ids), len(tasks); want != got {
		t.Fatalf("Expected %d tasks in 'all' query, got %d", want, got)
	}
	want := make(map[uuid.UUID]bool)
	for _, id := range ids {
		want[id] = true
	}
	for _, task := range tasks {
		if !want[task.ID] {
			t.Fatalf("Wanted queried task %d to have ID present in task listing, but not found", task.ID)
		}
	}
	idSubSet := []uuid.UUID{ids[1], ids[3]}
	tasks, err = client.Tasks(ctx, queue, entroq.WithTaskID(idSubSet...))
	if err != nil {
		t.Fatalf("Error getting tasks from queue %q: %v", queue, err)
	}
	if want, got := len(idSubSet), len(tasks); want != got {
		t.Fatalf("Expected %d tasks in 'all' query, got %d", want, got)
	}
	for i, task := range tasks {
		if want, got := idSubSet[i], task.ID; want != got {
			t.Fatalf("Wanted queried task %d to have ID %q, got %q", i, want, got)
		}
	}
}
TasksWithID exercises the task query mechanism that allows specific task IDs to be looked up.
package qtest
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"log"
	"net"
	"path"
	"strings"
	"testing"
	"time"
	"github.com/google/go-cmp/cmp"
	"github.com/google/uuid"
	"github.com/shiblon/entroq"
	"github.com/shiblon/entroq/backend/eqgrpc"
	"github.com/shiblon/entroq/qsvc"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/test/bufconn"
	pb "github.com/shiblon/entroq/proto"
	hpb "google.golang.org/grpc/health/grpc_health_v1"
)
const bufSize = 1 << 20
type Dialer func() (net.Conn, error)
type Tester func(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string)
func ClientService(ctx context.Context, opener entroq.BackendOpener) (client *entroq.EntroQ, stop func(), err error) {
	s, dial, err := StartService(ctx, opener)
	if err != nil {
		return nil, nil, fmt.Errorf("client service: %w", err)
	}
	defer func() {
		if err != nil {
			s.Stop()
		}
	}()
	client, err = entroq.New(ctx, eqgrpc.Opener("bufnet",
		eqgrpc.WithNiladicDialer(dial),
		eqgrpc.WithInsecure()))
	if err != nil {
		return nil, nil, fmt.Errorf("start client on in-memory service: %w", err)
	}
	return client, func() {
		client.Close()
		s.Stop()
	}, nil
}
func StartService(ctx context.Context, opener entroq.BackendOpener) (*grpc.Server, Dialer, error) {
	lis := bufconn.Listen(bufSize)
	svc, err := qsvc.New(ctx, opener)
	if err != nil {
		return nil, nil, fmt.Errorf("start service: %w", err)
	}
	s := grpc.NewServer()
	hpb.RegisterHealthServer(s, health.NewServer())
	pb.RegisterEntroQServer(s, svc)
	go s.Serve(lis)
	return s, lis.Dial, nil
}
func SimpleChange(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	inQueue := path.Join(qPrefix, "simple_change", "in")
	outQueue := path.Join(qPrefix, "simple_change", "out")
	inserted, _, err := client.Modify(ctx, entroq.InsertingInto(inQueue))
	if err != nil {
		t.Fatalf("Error inserting: %v", err)
	}
	_, changed, err := client.Modify(ctx, inserted[0].AsChange(entroq.QueueTo(outQueue)))
	if err != nil {
		t.Fatalf("Error changing: %v", err)
	}
	if changed[0].Queue != outQueue {
		t.Fatalf("Change queue: want %q, got %v", outQueue, changed[0].Queue)
	}
	changed[0].Queue = inQueue
	if diff := EqualTasksVersionIncr(inserted[0], changed[0], 1); diff != "" {
		t.Fatalf("Tasks not equal (except version bump):\n%v", diff)
	}
}
func SimpleWorker(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	queue := path.Join(qPrefix, "simple_worker")
	var consumed []*entroq.Task
	ctx, cancel := context.WithCancel(ctx)
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return client.NewWorker(queue).Run(ctx, func(ctx context.Context, task *entroq.Task) ([]entroq.ModifyArg, error) {
			if task.Claims != 1 {
				return nil, fmt.Errorf("worker claim expected claims to be 1, got %d", task.Claims)
			}
			consumed = append(consumed, task)
			return []entroq.ModifyArg{task.AsDeletion()}, nil
		})
	})
	select {
	case <-time.After(1 * time.Second):
	case <-ctx.Done():
		t.Fatalf("Sleep: %v", ctx.Err())
	}
	var inserted []*entroq.Task
	for i := 0; i < 10; i++ {
		ins, _, err := client.Modify(ctx, entroq.InsertingInto(queue))
		if err != nil {
			t.Fatalf("Failed to insert task: %v", err)
		}
		inserted = append(inserted, ins...)
		select {
		case <-ctx.Done():
			t.Fatalf("Canceled while inserting: %v", ctx.Err())
		default:
		}
	}
	for {
		empty, err := client.QueuesEmpty(ctx, entroq.MatchExact(queue))
		if err != nil {
			t.Fatalf("Error checking for empty queue: %v", err)
		}
		if empty {
			break
		}
		select {
		case <-ctx.Done():
			t.Fatalf("Context error waiting for queues to empty: %v", err)
		default:
		}
	}
	cancel()
	if err := g.Wait(); err != nil && !entroq.IsCanceled(err) {
		t.Fatalf("Worker exit error: %v", err)
	}
	if diff := EqualAllTasksVersionIncr(inserted, consumed, 1); diff != "" {
		t.Errorf("Tasks inserted not the same as tasks consumed:\n%v", diff)
	}
}
func MultiWorker(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	bigQueue := path.Join(qPrefix, "multi_worker_big")
	medQueue := path.Join(qPrefix, "multi_worker_medium")
	smallQueue := path.Join(qPrefix, "multi_worker_small")
	const (
		bigSize   = 300
		medSize   = 60
		smallSize = 20
		numWorkers = 5
	)
	for i := 0; i < bigSize; i++ {
		args := []entroq.ModifyArg{
			entroq.InsertingInto(bigQueue, entroq.WithValue([]byte("big value"))),
		}
		if i < medSize {
			args = append(args, entroq.ModifyArg(
				entroq.InsertingInto(medQueue, entroq.WithValue([]byte("med value"))),
			))
		}
		if i < smallSize {
			args = append(args, entroq.ModifyArg(
				entroq.InsertingInto(smallQueue, entroq.WithValue([]byte("smallvalue"))),
			))
		}
		if _, _, err := client.Modify(ctx, args...); err != nil {
			t.Fatalf("Insert queues failed: %v", err)
		}
	}
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	var consumed []*entroq.Task
	consumedCh := make(chan *entroq.Task)
	g, ctx := errgroup.WithContext(ctx)
	for i := 0; i < numWorkers; i++ {
		g.Go(func() error {
			ti := 0
			w := client.NewWorker(bigQueue, medQueue, smallQueue)
			err := w.Run(ctx, func(ctx context.Context, task *entroq.Task) ([]entroq.ModifyArg, error) {
				ti++
				if task.Claims != 1 {
					return nil, fmt.Errorf("worker claim expected to be 1, was %d", task.Claims)
				}
				consumedCh <- task
				return []entroq.ModifyArg{task.AsDeletion()}, nil
			})
			if entroq.IsCanceled(err) {
				return nil
			}
			return err
		})
	}
	g.Go(func() error {
		waitCtx, _ := context.WithTimeout(ctx, 1*time.Minute)
		if err := client.WaitQueuesEmpty(waitCtx, entroq.MatchExact(bigQueue, medQueue, smallQueue)); err != nil {
			return fmt.Errorf("waiting for empty queues: %w", err)
		}
		cancel()
		return nil
	})
	go func() {
		g.Wait()
		close(consumedCh)
	}()
	for task := range consumedCh {
		consumed = append(consumed, task)
	}
	if err := g.Wait(); err != nil && !entroq.IsCanceled(err) {
		t.Fatalf("Error in worker")
	}
	queuesFound := make(map[string]int)
	var smallIndices []int
	var medIndices []int
	for i, t := range consumed {
		queuesFound[t.Queue]++
		switch t.Queue {
		case medQueue:
			medIndices = append(medIndices, i)
		case smallQueue:
			smallIndices = append(smallIndices, i)
		}
	}
	sortedMedian := func(indices []int) float64 {
		if len(indices)%2 == 1 {
			return float64(indices[len(indices)/2])
		}
		return float64(indices[len(indices)/2-1]+indices[len(indices)/2]) / 2
	}
	smallMedian := sortedMedian(smallIndices)
	medMedian := sortedMedian(medIndices)
	if found := queuesFound[bigQueue]; found != bigSize {
		t.Errorf("Expected to consume %d from big queue, consumed %d", bigSize, found)
	}
	if found := queuesFound[medQueue]; found != medSize {
		t.Errorf("Expected to consume %d from med queue, consumed %d", medSize, found)
	}
	if found := queuesFound[smallQueue]; found != smallSize {
		t.Errorf("Expected to consume %d from small queue, consumed %d", smallSize, found)
	}
	const (
		maxExpectedSmallMedian = float64(smallSize*3/2 + smallSize)
		maxExpectedMedMedian   = float64((medSize-smallSize)*2+smallSize*3)/2 + medSize
	)
	if smallMedian > maxExpectedSmallMedian {
		t.Errorf("Expected small median to max out at around %f, but was %f", maxExpectedSmallMedian, smallMedian)
	}
	if medMedian > maxExpectedMedMedian {
		t.Errorf("Expected med median to max out at around %f, but was %f", maxExpectedMedMedian, medMedian)
	}
}
func WorkerDependencyHandler(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	queue := path.Join(qPrefix, "dep-handler-queue")
	timesHandled := make(chan int, 1)
	timesHandled <- 0
	upgradeError := errors.New("upgraded")
	leaseTime := 3 * time.Second
	onDep := func(err entroq.DependencyError) error {
		h := <-timesHandled
		defer func() {
			timesHandled <- h + 1
			time.Sleep(leaseTime)
		}()
		switch h {
		case 0:
			return nil  
		case 1:
			return upgradeError  
		default:
			t.Fatalf("Called dependency handler too many times: %v", timesHandled)
			return nil
		}
	}
	if _, _, err := client.Modify(ctx, entroq.InsertingInto(queue)); err != nil {
		t.Fatalf("Failed to insert task: %v", err)
	}
	w := client.NewWorker(queue).WithOpts(entroq.WithDependencyHandler(onDep), entroq.WithLease(leaseTime))
	ctx, cancel := context.WithTimeout(ctx, 3*leaseTime)
	defer cancel()
	err := w.Run(ctx, func(ctx context.Context, task *entroq.Task) ([]entroq.ModifyArg, error) {
		return []entroq.ModifyArg{task.AsDeletion(), entroq.DependingOn(uuid.New(), 0, entroq.WithIDQueue("no queue"))}, nil
	})
	if !errors.Is(err, upgradeError) {
		t.Fatalf("Expected upgrade error, got %v", err)
	}
	if h := <-timesHandled; h != 2 {
		t.Fatalf("Expected to have a safe error, then a fatal error, only ran %v times", h)
	}
}
func WorkerRetryOnError(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	newTask := func(val string) *entroq.Task {
		return &entroq.Task{
			Queue: path.Join(qPrefix, "retry_on_error", val),
			ID:    uuid.New(),
			Value: []byte(val),
		}
	}
	type tc struct {
		name        string
		input       *entroq.Task
		maxAttempt  int32
		wantAttempt int32
		wantErr     string
		wantMove    bool
	}
	cases := []tc{
		{
			name:        "retry-no-max",
			input:       newTask("retry"),
			maxAttempt:  0,
			wantAttempt: 1,
			wantErr:     "retry error: retry",
			wantMove:    false,
		},
		{
			name:        "retry-larger-max",
			input:       newTask("with max"),
			maxAttempt:  3,
			wantAttempt: 1,
			wantErr:     "retry error: with max",
			wantMove:    false,
		},
		{
			name:        "retry-too-many",
			input:       newTask("too many"),
			maxAttempt:  1,
			wantAttempt: 1,
			wantErr:     "retry error: too many",
			wantMove:    true,
		},
	}
	runWorkerOneCase := func(ctx context.Context, c tc) {
		t.Helper()
		w := client.NewWorker(c.input.Queue).WithOpts(entroq.WithBaseRetryDelay(0), entroq.WithMaxAttempts(c.maxAttempt))
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()
		retriedTaskCh := make(chan *entroq.Task, 1)
		g, gctx := errgroup.WithContext(ctx)
		g.Go(func() error {
			return w.Run(gctx, func(ctx context.Context, task *entroq.Task) ([]entroq.ModifyArg, error) {
				if task.Attempt == 0 {
					return nil, entroq.RetryTaskErrorf("retry error: %s", string(task.Value))
				}
				retriedTaskCh <- task
				return []entroq.ModifyArg{task.AsDeletion()}, nil
			})
		})
		if _, _, err := client.Modify(ctx, entroq.InsertingInto(c.input.Queue, entroq.WithID(c.input.ID), entroq.WithValue(c.input.Value))); err != nil {
			t.Fatalf("Test %q insert task: %v", c.name, err)
		}
		waitCtx, _ := context.WithTimeout(ctx, 5*time.Second)
		var changedTask *entroq.Task
		if c.wantMove {
			if err := client.WaitQueuesEmpty(waitCtx, entroq.MatchExact(c.input.Queue)); err != nil && !entroq.IsCanceled(err) {
				t.Fatalf("Test %q expected queue %q to become empty, didn't happen: %v", c.name, c.input.Queue, err)
			}
			errTasks, err := client.Tasks(ctx, w.ErrQMap(c.input.Queue))
			if err != nil {
				t.Fatalf("Test %q can't get tasks from error queue: %v", c.name, err)
			}
			if want, got := 1, len(errTasks); want != got {
				t.Fatalf("Test %q expected %d error tasks, got %d", c.name, want, got)
			}
			changedTask = errTasks[0]
		} else {
			select {
			case changedTask = <-retriedTaskCh:
				if want, got := c.input.ID, changedTask.ID; want != got {
					t.Fatalf("Test %q retry task expected ID %v, got %v", c.name, want, got)
				}
			case <-time.After(5 * time.Second):  
				t.Fatalf("Test %q took too long getting the retried task from something that had multiple attempts", c.name)
			}
		}
		if want, got := c.wantErr, changedTask.Err; want != got {
			t.Fatalf("Test %q expected err %q, got %q", c.name, want, got)
		}
		if want, got := c.wantAttempt, changedTask.Attempt; want != got {
			t.Fatalf("Test %q expected attempt %d, got %d", c.name, want, got)
		}
		cancel()
		if err := g.Wait(); err != nil && !entroq.IsCanceled(err) {
			t.Fatalf("Test %q failed worker wait after cancel: %v", c.name, err)
		}
	}
	for _, test := range cases {
		runWorkerOneCase(ctx, test)
	}
}
func WorkerMoveOnError(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	baseQueue := path.Join(qPrefix, "move_on_error")
	type tc struct {
		name  string
		input *entroq.Task
		die   bool
		moved bool
	}
	newTask := func(val string) *entroq.Task {
		id := uuid.New()
		return &entroq.Task{
			Queue: path.Join(baseQueue, val, id.String()),
			ID:    id,
			Value: []byte(val),
		}
	}
	cases := []tc{
		{
			name:  "die",
			input: newTask("die"),
			die:   true,
		},
		{
			name:  "move",
			input: newTask("move"),
			moved: true,
		},
		{
			name:  "wait-for-renewal",
			input: newTask("move-wait"),
			moved: true,
		},
	}
	runWorkerOneCase := func(ctx context.Context, c tc) {
		c.input.ID = uuid.New()
		const leaseTime = 5 * time.Second
		w := client.NewWorker(c.input.Queue).WithOpts(
			entroq.WithLease(leaseTime),
		)
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()
		g, gctx := errgroup.WithContext(ctx)
		g.Go(func() error {
			if err := w.Run(gctx, func(ctx context.Context, task *entroq.Task) ([]entroq.ModifyArg, error) {
				switch string(task.Value) {
				case "die":
					return nil, fmt.Errorf("task asked to die")
				case "move":
					return nil, entroq.MoveTaskErrorf("task asked to move")
				case "move-wait":
					select {
					case <-time.After(leaseTime):
						return nil, entroq.MoveTaskErrorf("task asked to move after renewal")
					case <-ctx.Done():
						return nil, fmt.Errorf("oops - test %q took too long, gave up before finishing: %w", c.name, ctx.Err())
					}
				default:
					return []entroq.ModifyArg{task.AsDeletion()}, nil
				}
			}); err != nil && !entroq.IsCanceled(err) {
				log.Printf("Worker Run error: %v", err)
				return err
			}
			return nil
		})
		if _, _, err := client.Modify(ctx, entroq.InsertingInto(c.input.Queue,
			entroq.WithID(c.input.ID),
			entroq.WithValue(c.input.Value),
		)); err != nil {
			t.Fatalf("Test %q insert task work: %v", c.name, err)
		}
		if c.die {
			if err := g.Wait(); err != nil && entroq.IsTimeout(err) {
				t.Fatalf("Test %q expected to die, but not with a timeout error: %v", c.name, err)
			}
			c.input.Version = 1
			if _, _, err := client.Modify(ctx, c.input.AsDeletion()); err != nil {
				t.Fatalf("Test %q tried to clean up dead task: %v", c.name, err)
			}
			return
		}
		waitCtx, _ := context.WithTimeout(ctx, 3*leaseTime)
		if err := client.WaitQueuesEmpty(waitCtx, entroq.MatchExact(c.input.Queue)); err != nil && !entroq.IsCanceled(err) {
			t.Fatalf("Test %q: no moved tasks found, task was not expected to die: %v", c.name, err)
		}
		errTasks, err := client.Tasks(ctx, w.ErrQMap(c.input.Queue))
		if err != nil {
			t.Fatalf("Test %q find in error queue: %v", c.name, err)
		}
		var foundTask *entroq.Task
		for _, task := range errTasks {
			if task.ID == c.input.ID {
				foundTask = task
				if !strings.Contains(foundTask.Err, "asked to move") {
					t.Fatalf("Test %q expected moved task to have a move error, had %q", c.name, foundTask.Err)
				}
				break
			}
		}
		if c.moved && foundTask == nil {
			t.Errorf("Test %q expected task to be moved, but is not found in %q", c.name, w.ErrQMap(c.input.Queue))
		} else if !c.moved && foundTask != nil {
			t.Errorf("Test %q expected task to be deleted, but showed up in %q", c.name, w.ErrQMap(c.input.Queue))
		}
		cancel()
		err = g.Wait()
		if c.moved && err != nil && !entroq.IsCanceled(err) {
			t.Errorf("Test %q expected no error on move, got %v", c.name, err)
		} else if !c.moved && entroq.IsCanceled(err) {
			t.Errorf("Test %q expected error from worker, got none", c.name)
		}
	}
	stressCount := 30
	if testing.Short() {
		stressCount = 1
	}
	for _, test := range cases {
		for i := 0; i < stressCount; i++ {
			t.Run(fmt.Sprintf("case=%v-%v", test.name, i), func(*testing.T) {
				runWorkerOneCase(ctx, test)
			})
		}
	}
}
func WorkerRenewal(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	queue := path.Join(qPrefix, "worker_renewal")
	_, _, err := client.Modify(ctx, entroq.InsertingInto(queue))
	if err != nil {
		t.Fatalf("Error inserting: %v", err)
	}
	task, err := client.Claim(ctx, entroq.From(queue), entroq.ClaimFor(6*time.Second))
	if err != nil {
		t.Fatalf("Failed to claim task: %v", err)
	}
	renewed, err := client.DoWithRenew(ctx, task, 6*time.Second, func(ctx context.Context) error {
		select {
		case <-ctx.Done():
			return fmt.Errorf("worker do with renew: %w", ctx.Err())
		case <-time.After(10 * time.Second):  
			return nil
		}
		return nil
	})
	if err != nil {
		t.Fatalf("Error renewing and waiting: %v", err)
	}
	if want, got := task.Version+3, renewed.Version; want != got {
		t.Fatalf("Expected renewed task to be at version %d, got %d", want, got)
	}
}
func TasksOmitValue(ctx context.Context, t *testing.T, client *entroq.EntroQ, qPrefix string) {
	queue := path.Join(qPrefix, "tasks_omit_value")
	inserted, _, err := client.Modify(ctx,
		entroq.InsertingInto(queue, entroq.WithValue([]byte("t1"))),
		entroq.InsertingInto(queue, entroq.WithValue([]byte("t2"))),
		entroq.InsertingInto(queue, entroq.WithValue([]byte("t3"))),
	)
	if err != nil {
		t.Fatalf("Failed to insert tasks: %v", err)
	}
	tasks, err := client.Tasks(ctx, queue, entroq.OmitValues())
	if err != nil {
		t.Fatalf("Failed to get tasks without values: %v", err)
	}
	if diff := EqualAllTasksOmitValues(inserted, tasks); diff != "" {
		t.Errorf("Task listing without values had unexpected results (-want +got):\n%v", diff)
	}
	tasksWithVals, err := client.Tasks(ctx, queue)
	if err != nil {
		t.Fatalf("Failed to get tasks with values: %v", err)
	}
	if diff := EqualAllTasks(inserted, tasksWithVals); diff != "" {
		t.Fatalf("Task listing with values had unexpected results (-want +got):\n%v", diff)
	}
}