helpers_test.go

// Copyright 2026 Hamed Yousefi
// SPDX-License-Identifier: MPL-2.0

package lockstep

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/hmdsefi/lockstep/internal/engine/enginetest"
	"github.com/hmdsefi/lockstep/internal/engine/local"
	"github.com/hmdsefi/lockstep/internal/supervise"
)

// newManualSystem returns a system built through the ACT-015 constructor with opts on a manual
// engine for node "local" with entropy seed 1 (ACT §10, "manual system").
func newManualSystem(t *testing.T, opts ...Option) (*System, *enginetest.Engine) {
	t.Helper()
	eng := enginetest.New("local", 1)
	sys, err := newSystem(eng.Engine(), opts)
	if err != nil {
		t.Fatalf("newSystem on a manual engine: %v", err)
	}
	return sys, eng
}

// newRecordingSystem returns a system built through the ACT-015 constructor with opts on a local
// engine whose tracer is a new recorder (ACT §10, "recording tracer"). The system shuts down when
// the test ends.
func newRecordingSystem(t *testing.T, opts ...Option) (*System, *enginetest.Recorder) {
	t.Helper()
	rec := enginetest.NewRecorder()
	le, err := local.New(local.Config{Node: "local", Tracer: rec})
	if err != nil {
		t.Fatalf("local.New: %v", err)
	}
	sys, err := newSystem(le.Engine(), opts)
	if err != nil {
		le.Stop()
		t.Fatalf("newSystem on a local engine: %v", err)
	}
	shutdownAtEnd(t, sys)
	return sys, rec
}

// newLocalSystem returns New(opts...) and shuts it down when the test ends.
func newLocalSystem(t *testing.T, opts ...Option) *System {
	t.Helper()
	sys := New(opts...)
	shutdownAtEnd(t, sys)
	return sys
}

// shutdownAtEnd shuts sys down when the test ends, waiting up to 30 s, so goleak finds no
// goroutine of its engine. A system the test shut down already returns at once. The bound is 30 s,
// not the usual 10 s, because some tests stop thousands of actors, under -race with coverage on a
// busy machine.
func shutdownAtEnd(t *testing.T, sys *System) {
	t.Helper()
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		if err := sys.Shutdown(ctx); err != nil {
			t.Errorf("Shutdown at the end of the test: %v", err)
		}
	})
}

// probe is ACT §10's test actor. It answers a Query, sent by inProbe through Ask, with the result
// of the Query's function run inside its Receive, so tests can observe ctx from inside a turn. It
// passes every other message to receive, if set, and forwards every signal it gets to signals.
type probe[M any] struct {
	receive func(ctx *Context[M], msg M)
	signals chan Signal // buffered (probeSignals); a full channel blocks the probe's turn
}

// probeSignals is the capacity of a probe's signal channel.
const probeSignals = 64

// Query is the message inProbe sends a probe: run fn during the probe's turn and answer reply
// with its result.
type Query[M any] struct {
	fn    func(ctx *Context[M]) any
	reply ReplyTo[any]
}

// Receive answers a Query and hands every other message to p.receive.
func (p *probe[M]) Receive(ctx *Context[M], msg M) {
	if q, ok := any(msg).(Query[M]); ok {
		q.reply.Send(ctx, q.fn(ctx))
		return
	}
	if p.receive != nil {
		p.receive(ctx, msg)
	}
}

// Signal forwards sig to p.signals (SignalReceiver).
func (p *probe[M]) Signal(_ *Context[M], sig Signal) { p.signals <- sig }

// spawnProbe spawns a probe named name under parent; receive, which may be nil, gets the messages
// that are not a Query. Every incarnation of the actor is the same probe value.
func spawnProbe[M any](t *testing.T, parent Spawner, name string, receive func(*Context[M], M)) (PID[M], *probe[M]) {
	t.Helper()
	p := &probe[M]{receive: receive, signals: make(chan Signal, probeSignals)}
	pid, err := Spawn(parent, name, actorOf[M](p))
	if err != nil {
		t.Fatalf("Spawn probe %q: %v", name, err)
	}
	return pid, p
}

// inProbe runs fn during a turn of the probe pid and returns fn's result. It asks the probe from
// sys and awaits the answer: on a local engine for up to 10 s, on a manual engine by running it
// until idle. M must be able to hold a Query[M] (for example any).
func inProbe[M any](t *testing.T, sys *System, pid PID[M], fn func(ctx *Context[M]) any) any {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	r := Ask(sys, pid, func(rt ReplyTo[any]) M {
		m, ok := any(Query[M]{fn: fn, reply: rt}).(M)
		if !ok {
			panic(fmt.Sprintf("inProbe: message type %T cannot hold a Query", m))
		}
		return m
	}, WithTimeout(10*time.Second))
	v, err := r.Await(ctx)
	if err != nil {
		t.Fatalf("inProbe(%v): %v", pid, err)
	}
	return v
}

// stepUntilSuspended runs the engine's turns one at a time until c is Suspended, so the test sees
// a failed cell before its supervisor's turn decides the failure. It fails the test if the engine
// runs out of turns first.
func stepUntilSuspended[M any](t *testing.T, eng *enginetest.Engine, c *cell[M]) {
	t.Helper()
	for {
		if st, _ := c.lifecycle(); st == supervise.Suspended {
			return
		}
		if !eng.Step() {
			t.Fatalf("%v never became suspended", c.addr)
		}
	}
}

// entries returns the number of entries in t's maps: asks of this system that are pending or
// tracked, and tracked asks of other nodes.
func (t *askTable) entries() (local, remote int) {
	for i := range t.shards {
		sh := &t.shards[i]
		sh.mu.Lock()
		local += len(sh.local)
		remote += len(sh.remote)
		sh.mu.Unlock()
	}
	return local, remote
}