hardlink/internal/dispenser/delivery_test.go

package dispenser

import (
	"context"
	"errors"
	"io"
	"reflect"
	"testing"
	"time"
)

func apReply(st []byte) []byte {
	frame := []byte{STX, 0x30, 0x30, 0, byte(len(st) + 2), 'S', 'F'}
	frame = append(frame, st...)
	frame = append(frame, ETX)
	return append(frame, calculateBCC(frame))
}

func workerRequest(c *Client, ctx context.Context, typ cmdType) cmdResp {
	ch := make(chan cmdResp, 1)
	c.handle(cmdReq{typ: typ, ctx: ctx, respCh: ch})
	return <-ch
}

func TestDeliveryDispatchBoundary(t *testing.T) {
	transportAddress(t)
	for _, tc := range []struct {
		name        string
		short       int
		badACK      bool
		cancelAfter int
		pending     bool
		wantErr     bool
	}{
		{name: "success", pending: true},
		{name: "command short write", short: 1, wantErr: true},
		{name: "bad ACK", badACK: true, wantErr: true},
		{name: "ENQ short write", short: 2, pending: true, wantErr: true},
		{name: "cancel before ENQ", cancelAfter: 1, wantErr: true},
		{name: "cancel after ENQ", cancelAfter: 2, pending: true, wantErr: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			ack := append([]byte(nil), vendorACK...)
			if tc.badACK {
				ack[0] = 0
			}
			p := &scriptedTransport{chunks: [][]byte{ack}, shortWrite: tc.short}
			p.afterWrite = func() {
				if len(p.writes) == tc.cancelAfter {
					cancel()
				}
			}
			c := &Client{port: p}
			r := workerRequest(c, ctx, cmdOutOfMouth)
			if (r.err != nil) != tc.wantErr || c.deliveryPending != tc.pending {
				t.Errorf("FC0(%s): err=%v pending=%v, want error=%v pending=%v", tc.name, r.err, c.deliveryPending, tc.wantErr, tc.pending)
			}
			if len(p.writes) > 2 {
				t.Errorf("FC0 writes=%d, want no resend", len(p.writes))
			}
		})
	}
	t.Run("definite failure preserves previous pending", func(t *testing.T) {
		c := &Client{port: &scriptedTransport{writeErr: io.ErrClosedPipe}, deliveryPending: true}
		workerRequest(c, context.Background(), cmdOutOfMouth)
		if !c.deliveryPending {
			t.Error("failed FC0 erased previous pending")
		}
	})
	t.Run("ENQ transport error remains pending", func(t *testing.T) {
		p := &scriptedTransport{chunks: [][]byte{vendorACK}}
		p.afterWrite = func() {
			if len(p.writes) == 2 {
				p.writeErr = io.ErrClosedPipe
			}
		}
		c := &Client{port: p}
		r := workerRequest(c, context.Background(), cmdOutOfMouth)
		if !errors.Is(r.err, io.ErrClosedPipe) || !c.deliveryPending {
			t.Errorf("ENQ error=%v pending=%v, want closed pipe and pending", r.err, c.deliveryPending)
		}
	})
}

func TestWorkerClearanceValidation(t *testing.T) {
	transportAddress(t)
	for _, tc := range []struct {
		name  string
		st    []byte
		clear bool
	}{
		{"clear", status(0x30), true},
		{"low stock clear", []byte{0x30, 0x30, 0x31, 0x30}, true},
		{"residual encoder", status(0x33), false},
		{"all sensors", status(0x37), false},
		{"mouth", status(0x34), false},
		{"unknown diagnostic", []byte{0x30, 0x30, 0xFF, 0x30}, false},
		{"jam", []byte{0x30, 0x30, 0x32, 0x30}, false},
		{"overlap", []byte{0x30, 0x30, 0x34, 0x30}, false},
		{"rejection", []byte{0x36, 0x30, 0x30, 0x30}, false},
		{"empty", status(0x38), false},
		{"short payload", []byte{0x30, 0x30, 0x30}, false},
		{"unknown position", status(0x40), false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			p := &scriptedTransport{chunks: [][]byte{vendorACK, apReply(tc.st), vendorACK}}
			c := &Client{port: p, deliveryPending: true, deliveryStarted: time.Now(), lastStatus: status(0x30), lastStatusT: time.Now(), statusTTL: time.Hour}
			r := workerRequest(c, context.Background(), cmdToEncoder)
			if (r.err == nil) != tc.clear || c.deliveryPending == tc.clear {
				t.Errorf("FC7(% X): err=%v pending=%v, want clear=%v", tc.st, r.err, c.deliveryPending, tc.clear)
			}
			want := []string{"AP"}
			if tc.clear {
				want = append(want, "FC7")
			}
			if got := wireCommands(p); !reflect.DeepEqual(got, want) {
				t.Errorf("commands=%v, want %v", got, want)
			}
		})
	}
	t.Run("bad BCC cannot clear", func(t *testing.T) {
		frame := apReply(status(0x30))
		frame[len(frame)-1] ^= 1
		p := &scriptedTransport{chunks: [][]byte{vendorACK, frame}}
		c := &Client{port: p, deliveryPending: true}
		r := workerRequest(c, context.Background(), cmdToEncoder)
		if r.err == nil || !c.deliveryPending {
			t.Errorf("bad BCC: err=%v pending=%v", r.err, c.deliveryPending)
		}
	})
}

func wireCommands(p *scriptedTransport) []string {
	var commands []string
	for _, w := range p.writes {
		if len(w) > 6 && w[0] == STX {
			commands = append(commands, string(w[5:len(w)-2]))
		}
	}
	return commands
}

func deliveryWorkerClient(t *testing.T, p *scriptedTransport) (*Client, *time.Time) {
	t.Helper()
	transportAddress(t)
	now := time.Unix(0, 0)
	c := &Client{port: p, reqCh: make(chan cmdReq, 16), done: make(chan struct{})}
	c.sequenceTiming = sequenceTiming{now: func() time.Time { return now }, wait: func(ctx context.Context, d time.Duration) error {
		if err := ctx.Err(); err != nil {
			return err
		}
		now = now.Add(d)
		return nil
	}}
	stopped := make(chan struct{})
	go func() { defer close(stopped); c.loop() }()
	t.Cleanup(func() { c.Close(); <-stopped })
	return c, &now
}

func TestDeliveryIncidentReplay(t *testing.T) {
	for _, next := range []bool{false, true} {
		t.Run(map[bool]string{false: "current", true: "next"}[next], func(t *testing.T) {
			p := &scriptedTransport{chunks: [][]byte{
				vendorACK, apReply(status(0x33)), // current card at encoder; encoding succeeds
				vendorACK,                        // FC0
				vendorACK, apReply(status(0x37)), // best effort must defer
				vendorACK, apReply(status(0x33)), // previous encoder sensor must not succeed
				vendorACK, apReply(status(0x34)),
				vendorACK, apReply(status(0x37)),
				vendorACK, apReply(status(0x30)), // clearance
				vendorACK, // FC7
				vendorACK, apReply(status(0x33)),
			}}
			c, now := deliveryWorkerClient(t, p)
			if _, err := c.PrepareCurrentCard(context.Background()); err != nil {
				t.Fatal(err)
			}
			if _, err := c.DeliverCurrentCard(context.Background()); err != nil {
				t.Fatal(err)
			}
			// A cached clear result must not authorize FC7.
			c.mu.Lock()
			c.lastStatus = status(0x30)
			c.lastStatusT = time.Now()
			c.statusTTL = time.Hour
			c.mu.Unlock()
			if err := c.BeginPrepareNextCard(context.Background()); !errors.Is(err, errDeliveryPending) {
				t.Fatalf("BeginPrepareNextCard=%v, want deferred", err)
			}
			if got := wireCommands(p); !reflect.DeepEqual(got, []string{"AP", "FC0", "AP"}) {
				t.Fatalf("before clearance commands=%v", got)
			}
			prepare := c.PrepareCurrentCard
			if next {
				prepare = c.PrepareNextCard
			}
			if _, err := prepare(context.Background()); err != nil {
				t.Fatal(err)
			}
			want := []string{"AP", "FC0", "AP", "AP", "AP", "AP", "AP", "FC7", "AP"}
			if got := wireCommands(p); !reflect.DeepEqual(got, want) {
				t.Errorf("incident commands=%v, want %v", got, want)
			}
			if elapsed := now.Sub(time.Unix(0, 0)); elapsed != 3*time.Second {
				t.Errorf("clearance waits=%s, want 3s", elapsed)
			}
		})
	}
}

func TestDeliveryClearImmediately(t *testing.T) {
	p := &scriptedTransport{chunks: [][]byte{vendorACK, vendorACK, apReply(status(0x30)), vendorACK, vendorACK, apReply(status(0x33))}}
	c, now := deliveryWorkerClient(t, p)
	if _, err := c.DeliverCurrentCard(context.Background()); err != nil {
		t.Fatal(err)
	}
	if _, err := c.PrepareCurrentCard(context.Background()); err != nil {
		t.Fatal(err)
	}
	if got := wireCommands(p); !reflect.DeepEqual(got, []string{"FC0", "AP", "FC7", "AP"}) {
		t.Errorf("commands=%v, want FC0 AP FC7 AP", got)
	}
	if !now.Equal(time.Unix(0, 0)) {
		t.Errorf("immediate clearance waited until %v", now)
	}
}

func TestDeliveryClearanceWaitStops(t *testing.T) {
	for _, tc := range []struct {
		name     string
		response cmdResp
		cancel   bool
		timeout  bool
	}{
		{name: "cancel", response: cmdResp{status: status(0x37), deliveryPending: true}, cancel: true},
		{name: "timeout", response: cmdResp{status: status(0x37), deliveryPending: true}, timeout: true},
		{name: "transport", response: cmdResp{err: io.ErrUnexpectedEOF}},
		{name: "jam", response: cmdResp{status: []byte{0x30, 0x30, 0x32, 0x30}, deliveryPending: true}},
		{name: "empty", response: cmdResp{status: status(0x38), deliveryPending: true}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			responses := make([]cmdResp, 20)
			for i := range responses {
				responses[i] = tc.response
			}
			c, d := newSequenceTestClient(t, responses...)
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			if tc.cancel {
				c.sequenceTiming.wait = func(context.Context, time.Duration) error { cancel(); return ctx.Err() }
			}
			started := c.sequenceTiming.now()
			_, err := c.PrepareCurrentCard(ctx)
			if err == nil {
				t.Fatal("preparation succeeded while clearance unavailable")
			}
			if tc.cancel && !errors.Is(err, context.Canceled) {
				t.Errorf("error=%v, want canceled", err)
			}
			if tc.timeout {
				if !errors.Is(err, context.DeadlineExceeded) || c.sequenceTiming.now().Sub(started) != sequenceTimeout {
					t.Errorf("timeout err=%v elapsed=%s", err, c.sequenceTiming.now().Sub(started))
				}
			}
			for _, cmd := range d.commands {
				if cmd != cmdStatus {
					t.Errorf("clearance dispatched command %v, want only AP", cmd)
				}
			}
		})
	}
}

func TestDeliveryClearanceDoesNotConsumeRecoveryWindow(t *testing.T) {
	responses := []cmdResp{{status: status(0x37), deliveryPending: true}, {status: status(0x33), deliveryPending: true}, {status: status(0x30)}}
	responses = append(responses, prepareFailureResponses(0x32, 7)...)
	responses = append(responses, cmdResp{status: status(0x30)}, cmdResp{status: status(0x33)})
	c, d := newSequenceTestClient(t, responses...)
	if _, err := c.PrepareCurrentCard(context.Background()); err != nil {
		t.Fatal(err)
	}
	var initial, reset time.Time
	for i, cmd := range d.commands {
		if cmd == cmdToEncoder && initial.IsZero() {
			initial = d.commandTimes[i]
		}
		if cmd == cmdReset {
			reset = d.commandTimes[i]
		}
	}
	if reset.Sub(initial) != sequenceRetryAfter || commandCount(d.commands, cmdReset) != 1 || commandCount(d.commands, cmdToEncoder) != 2 {
		t.Errorf("recovery commands=%v reset delay=%s, want two FC7 and one RS at 6s", d.commands, reset.Sub(initial))
	}
	if got := c.sequenceTiming.now().Sub(reset); got != sequenceResetWait+sequencePollInterval {
		t.Errorf("post-reset elapsed=%s, want settle plus poll=3s", got)
	}
}

func TestDeliveryPendingIsNotPersisted(t *testing.T) {
	c := NewClient(nil, 1)
	defer c.Close()
	if c.deliveryPending {
		t.Error("new Client has pending delivery")
	}
	p := &scriptedTransport{chunks: [][]byte{vendorACK, apReply(status(0x37))}}
	fresh, _ := deliveryWorkerClient(t, p)
	if _, err := fresh.PrepareCurrentCard(context.Background()); err != nil {
		t.Errorf("fresh client encoder status: %v", err)
	}
	if got := wireCommands(p); !reflect.DeepEqual(got, []string{"AP"}) {
		t.Errorf("restart commands=%v, want AP only", got)
	}
}

func TestDeliveryCancellationAfterACKDoesNotSetPending(t *testing.T) {
	transportAddress(t)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	p := &scriptedTransport{chunks: [][]byte{vendorACK}, afterRead: cancel}
	c := &Client{port: p}
	r := workerRequest(c, ctx, cmdOutOfMouth)
	if !errors.Is(r.err, context.Canceled) || c.deliveryPending || len(p.writes) != 1 {
		t.Errorf("cancel after ACK: err=%v pending=%v writes=%d, want canceled, false, 1", r.err, c.deliveryPending, len(p.writes))
	}
}

func TestDeliveryClearanceRetainsFullPreparationTimeout(t *testing.T) {
	responses := make([]cmdResp, 8)
	for i := range responses {
		responses[i] = cmdResp{status: status(0x37), deliveryPending: true}
	}
	responses = append(responses, cmdResp{status: status(0x30)})
	responses = append(responses, prepareFailureResponses(0x32, 30)...)
	c, d := newSequenceTestClient(t, responses...)
	started := c.sequenceTiming.now()
	if _, err := c.PrepareCurrentCard(context.Background()); err == nil {
		t.Fatal("persistent failure succeeded")
	}
	if got := c.sequenceTiming.now().Sub(started); got != 8*time.Second+sequenceTimeout {
		t.Errorf("total duration=%s, want clearance 8s + preparation 16s", got)
	}
	if commandCount(d.commands, cmdReset) != 1 || commandCount(d.commands, cmdToEncoder) != 2 {
		t.Errorf("commands=%v, want one RS and two FC7", d.commands)
	}
}

func TestWorkerSerializesClearanceAndFC7(t *testing.T) {
	p := &scriptedTransport{chunks: [][]byte{vendorACK, apReply(status(0x30)), vendorACK, vendorACK, vendorACK, apReply(status(0x37))}}
	c, _ := deliveryWorkerClient(t, p)
	// Set initial state before any request can reach the worker.
	c.deliveryPending = true
	c.deliveryStarted = time.Now()
	reading := make(chan struct{})
	resume := make(chan struct{})
	p.afterWrite = func() {
		if len(p.writes) == 1 {
			close(reading)
			<-resume
		}
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	fc7 := make(chan error, 1)
	go func() { fc7 <- c.BeginPrepareNextCard(ctx) }()
	select {
	case <-reading:
	case <-ctx.Done():
		t.Fatal("worker did not begin AP")
	}
	fc0 := make(chan cmdResp, 1)
	// This FC0 is queued while AP is in progress. It must not slip between
	// the clearance observation and FC7 dispatch.
	c.reqCh <- cmdReq{typ: cmdOutOfMouth, ctx: ctx, respCh: fc0}
	close(resume)
	if err := <-fc7; err != nil {
		t.Fatal(err)
	}
	if r := <-fc0; r.err != nil {
		t.Fatal(r.err)
	}
	if err := c.ToEncoder(ctx); !errors.Is(err, errDeliveryPending) {
		t.Errorf("FC7 after queued FC0=%v, want deferred", err)
	}
	want := []string{"AP", "FC7", "FC0", "AP"}
	if got := wireCommands(p); !reflect.DeepEqual(got, want) {
		t.Errorf("serialized commands=%v, want %v", got, want)
	}
}