bldr/resource/cross-runtime_test.go
Rama master · réplica de lectura del repositorio original.
package resource_test
import (
"bufio"
"bytes"
"context"
"encoding/binary"
"errors"
"io"
"net"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/aperturerobotics/starpc/srpc"
"github.com/s4wave/spacewave/bldr/resource"
resource_client "github.com/s4wave/spacewave/bldr/resource/client"
resource_server "github.com/s4wave/spacewave/bldr/resource/server"
)
const crossRuntimeFixtureTimeout = 20 * time.Second
type crossRuntimeProcess struct {
cmd *exec.Cmd
input io.WriteCloser
// outputMtx guards output and stderr, which os/exec and the stdout
// scanner write while failure reporting reads them.
outputMtx sync.Mutex
output []string
stderr bytes.Buffer
lines chan string
done chan struct{}
err error
}
func startCrossRuntimeProcess(
t *testing.T,
ctx context.Context,
name string,
args ...string,
) *crossRuntimeProcess {
// Resolve the repository root and build the subprocess command.
t.Helper()
root, err := filepath.Abs("../..")
if err != nil {
t.Fatalf("resolve repository root: %v", err)
}
// Without node_modules, bun auto-installs packages outside the lockfile,
// and the fixtures fail inside mismatched generated code.
if name == "bun" {
if _, err := os.Stat(filepath.Join(root, "node_modules")); err != nil {
t.Fatalf("bun fixtures need the lockfile's packages: run bun install --frozen-lockfile in %s", root)
}
}
command := exec.CommandContext(ctx, name, args...)
command.Dir = root
command.Env = os.Environ()
// Open the subprocess stdin and stdout pipes.
input, err := command.StdinPipe()
if err != nil {
t.Fatalf("%s stdin: %v", name, err)
}
stdout, err := command.StdoutPipe()
if err != nil {
t.Fatalf("%s stdout: %v", name, err)
}
// Start the subprocess with stderr captured under the output lock.
process := &crossRuntimeProcess{
cmd: command,
input: input,
lines: make(chan string, 64),
done: make(chan struct{}),
}
command.Stderr = crossRuntimeStderr{process: process}
if err := command.Start(); err != nil {
t.Fatalf("start %s: %v", name, err)
}
// Scan stdout lines into the process output and line channel.
scanDone := make(chan struct{})
go func() {
defer close(scanDone)
scanner := bufio.NewScanner(stdout)
for scanner.Scan() {
line := scanner.Text()
process.outputMtx.Lock()
process.output = append(process.output, line)
process.outputMtx.Unlock()
process.lines <- line
}
}()
// Wait for the subprocess and close the done channel after scanning ends.
go func() {
process.err = command.Wait()
<-scanDone
close(process.done)
}()
return process
}
// crossRuntimeStderr appends subprocess stderr under the output lock.
type crossRuntimeStderr struct {
process *crossRuntimeProcess
}
// Write appends one stderr chunk.
func (w crossRuntimeStderr) Write(data []byte) (int, error) {
w.process.outputMtx.Lock()
defer w.process.outputMtx.Unlock()
return w.process.stderr.Write(data)
}
func (p *crossRuntimeProcess) send(t *testing.T, command string) {
t.Helper()
if _, err := io.WriteString(p.input, command+"\n"); err != nil {
t.Fatalf("send %q to %s: %v\n%s", command, p.cmd.Path, err, p.logs())
}
}
func (p *crossRuntimeProcess) waitLine(
t *testing.T,
ctx context.Context,
want string,
) string {
t.Helper()
for {
select {
case line := <-p.lines:
if line == want || strings.HasPrefix(line, want) {
return line
}
case <-p.done:
t.Fatalf("%s exited before %q: %v\n%s", p.cmd.Path, want, p.err, p.logs())
case <-ctx.Done():
t.Fatalf("timed out waiting for %q from %s\n%s", want, p.cmd.Path, p.logs())
}
}
}
func (p *crossRuntimeProcess) waitExit(t *testing.T, ctx context.Context) {
t.Helper()
select {
case <-p.done:
if p.err != nil {
t.Fatalf("%s failed: %v\n%s", p.cmd.Path, p.err, p.logs())
}
case <-ctx.Done():
t.Fatalf("timed out waiting for %s\n%s", p.cmd.Path, p.logs())
}
}
func (p *crossRuntimeProcess) logs() string {
// Join the captured stdout and stderr under the output lock.
p.outputMtx.Lock()
defer p.outputMtx.Unlock()
logs := strings.Join(p.output, "\n")
if p.stderr.Len() != 0 {
logs += "\n" + p.stderr.String()
}
return logs
}
func TestPythonClientResourceLifecycleAgainstTypeScriptServer(t *testing.T) {
// Start the TypeScript server subprocess with a bounded context.
ctx, cancel := context.WithTimeout(t.Context(), crossRuntimeFixtureTimeout)
defer cancel()
server := startCrossRuntimeProcess(
t,
ctx,
"bun",
"run",
"./bldr/resource/testdata/cross-runtime-ts-core-server.ts",
"127.0.0.1:0",
)
ready := server.waitLine(t, ctx, "READY ")
address := strings.TrimPrefix(ready, "READY ")
client := startCrossRuntimeProcess(
t,
ctx,
"uv",
"run",
"python",
"bldr/resource/testdata/cross-runtime-python-client.py",
address,
)
// Drive the Python client through adopt and release against the TypeScript server.
server.waitLine(t, ctx, "ADOPT_ACK_HELD")
server.send(t, "ALLOW_ADOPT")
server.waitLine(t, ctx, "HANDLER_FINALLY")
server.waitLine(t, ctx, "RELEASE_COMPLETE")
server.waitLine(t, ctx, "HANDLER_FINALLY")
// Invalidate the TypeScript server and wait for both runtimes to drain.
client.waitLine(t, ctx, "PY_CLIENT_READY_TO_INVALIDATE")
server.send(t, "INVALIDATE")
server.waitLine(t, ctx, "TS_SERVER_OWNER_ZERO")
client.waitLine(t, ctx, "PY_CLIENT_OWNER_ZERO")
client.waitExit(t, ctx)
server.waitExit(t, ctx)
}
func TestTypeScriptClientResourceLifecycleAgainstPythonServer(t *testing.T) {
// Start the Python server subprocess with a bounded context.
ctx, cancel := context.WithTimeout(t.Context(), crossRuntimeFixtureTimeout)
defer cancel()
server := startCrossRuntimeProcess(
t,
ctx,
"uv",
"run",
"python",
"bldr/resource/testdata/cross-runtime-python-server.py",
"--listen",
"127.0.0.1:0",
)
// Start the TypeScript client subprocess against the Python server address.
ready := server.waitLine(t, ctx, "READY ")
client := startCrossRuntimeProcess(
t,
ctx,
"bun",
"run",
"./bldr/resource/testdata/cross-runtime-ts-core-client.ts",
strings.TrimPrefix(ready, "READY "),
)
// Drive the TypeScript client through adopt and release against the Python server.
server.waitLine(t, ctx, "ADOPT_ACK_HELD")
server.send(t, "ALLOW_ADOPT")
server.waitLine(t, ctx, "HANDLER_FINALLY")
server.waitLine(t, ctx, "HANDLER_FINALLY")
server.waitLine(t, ctx, "RELEASE_ENTERED")
server.send(t, "ALLOW_RELEASE")
server.waitLine(t, ctx, "RELEASE_COMPLETE")
// Invalidate the Python server and wait for both runtimes to drain.
client.waitLine(t, ctx, "TS_CLIENT_READY_TO_INVALIDATE")
server.send(t, "INVALIDATE")
server.waitLine(t, ctx, "PY_SERVER_OWNER_ZERO")
client.waitLine(t, ctx, "TS_CLIENT_OWNER_ZERO")
client.waitExit(t, ctx)
server.waitExit(t, ctx)
}
func TestGoClientResourceLifecycleAgainstPythonServer(t *testing.T) {
// Start the Python server subprocess with a bounded context.
ctx, cancel := context.WithTimeout(t.Context(), crossRuntimeFixtureTimeout)
defer cancel()
server := startCrossRuntimeProcess(
t,
ctx,
"uv",
"run",
"python",
"bldr/resource/testdata/cross-runtime-python-server.py",
"--listen",
"127.0.0.1:0",
)
// Build the Go ResourceClient against the Python server listener.
ready := server.waitLine(t, ctx, "READY ")
service := resource.NewSRPCResourceServiceClient(
newCrossRuntimeTCPClient(strings.TrimPrefix(ready, "READY ")),
)
client, err := resource_client.NewClient(ctx, service)
if err != nil {
t.Fatalf("start Go ResourceClient: %v", err)
}
root := client.AccessRootResource()
rootClient, err := root.GetClient()
if err != nil {
t.Fatalf("get Go root client: %v", err)
}
// Open the Go root resource and start the nested Spawn call in the background.
spawnResult := make(chan struct {
data []byte
err error
}, 1)
go func() {
data, callErr := crossRuntimeCall(
ctx,
rootClient,
"test.Root",
"Spawn",
[]byte("spawn"),
)
spawnResult <- struct {
data []byte
err error
}{data, callErr}
}()
// Hold the delayed acknowledgement and confirm the Spawn call stays blocked.
server.waitLine(t, ctx, "ADOPT_ACK_HELD")
select {
case result := <-spawnResult:
t.Fatalf("Go nested route opened before delayed acknowledgement: %q/%v", result.data, result.err)
default:
}
// Allow the adopt and collect the spawned child reference.
server.send(t, "ALLOW_ADOPT")
var childData []byte
select {
case result := <-spawnResult:
if result.err != nil {
t.Fatalf("spawn Python child: %v", result.err)
}
childData = result.data
case <-ctx.Done():
t.Fatal("timed out opening Go root route after acknowledgement")
}
// Validate the spawned child ID length and value.
if len(childData) != 4 {
t.Fatalf("child ID length = %d, want 4", len(childData))
}
childID := binary.BigEndian.Uint32(childData)
if childID == 0 {
t.Fatal("Spawn returned an empty child ID")
}
// Stream through the Python child resource and check the response.
child := client.CreateResourceReference(childID)
childClient, err := child.GetClient()
if err != nil {
t.Fatalf("get child client: %v", err)
}
data, err := crossRuntimeCall(ctx, childClient, "test.Child", "Stream", []byte("stream"))
if err != nil {
t.Fatalf("stream Python child: %v", err)
}
if string(data) != "later" {
t.Fatalf("stream data = %q, want later", data)
}
// Open a cancellable child stream and verify the active data before canceling it.
canceled, err := childClient.NewStream(
ctx,
"test.Child",
"Block",
srpc.NewRawMessage([]byte("block"), true),
)
if err != nil {
t.Fatalf("open cancellable child stream: %v", err)
}
active := srpc.NewRawMessage(nil, true)
if err := canceled.MsgRecv(active); err != nil {
t.Fatalf("receive cancellable child data: %v", err)
}
if string(active.GetData()) != "active" {
t.Fatalf("cancellable data = %q, want active", active.GetData())
}
if err := canceled.Close(); err != nil {
t.Fatalf("cancel child stream: %v", err)
}
server.waitLine(t, ctx, "HANDLER_FINALLY")
// Open a second child stream and release the child while it is active.
released, err := childClient.NewStream(
ctx,
"test.Child",
"Block",
srpc.NewRawMessage([]byte("block"), true),
)
if err != nil {
t.Fatalf("open releasable child stream: %v", err)
}
if err := released.MsgRecv(active); err != nil {
t.Fatalf("receive releasable child data: %v", err)
}
child.Release()
server.waitLine(t, ctx, "HANDLER_FINALLY")
server.waitLine(t, ctx, "RELEASE_ENTERED")
// Start a root Echo call that must stay blocked behind the child release barrier.
echoResult := make(chan struct {
data []byte
err error
}, 1)
go func() {
data, callErr := crossRuntimeCall(
ctx,
rootClient,
"test.Root",
"Echo",
[]byte("after-release"),
)
echoResult <- struct {
data []byte
err error
}{data, callErr}
}()
select {
case result := <-echoResult:
t.Fatalf("root route passed the child release barrier: %q/%v", result.data, result.err)
default:
}
// Release the child barrier and verify the root Echo completes afterwards.
server.send(t, "ALLOW_RELEASE")
server.waitLine(t, ctx, "RELEASE_COMPLETE")
select {
case result := <-echoResult:
if result.err != nil {
t.Fatalf("route root after child release: %v", result.err)
}
if string(result.data) != "after-release" {
t.Fatalf("root response = %q, want after-release", result.data)
}
case <-ctx.Done():
t.Fatal("timed out routing root after child release")
}
if err := released.Close(); err != nil && !errors.Is(err, context.Canceled) {
t.Fatalf("close released child stream: %v", err)
}
// Release the root and verify a fresh reference reuses the retained route.
root.Release()
reused := client.AccessRootResource()
reusedClient, err := reused.GetClient()
if err != nil {
t.Fatalf("get retained root client: %v", err)
}
data, err = crossRuntimeCall(ctx, reusedClient, "test.Root", "Echo", []byte("reused"))
if err != nil {
t.Fatalf("route retained root: %v", err)
}
if string(data) != "reused" {
t.Fatalf("retained root response = %q, want reused", data)
}
reused.Release()
// Invalidate the server generation and confirm the client and stale reference fail.
server.send(t, "INVALIDATE")
select {
case <-client.Done():
case <-ctx.Done():
t.Fatal("timed out invalidating Go ResourceClient generation")
}
stale := client.AccessRootResource() //nolint:lostresource // The server has invalidated this generation; the test verifies the returned reference is already released.
if _, err := stale.GetClient(); !errors.Is(err, resource.ErrResourceOrClientReleased) {
t.Fatalf("invalidated root error = %v, want %v", err, resource.ErrResourceOrClientReleased)
}
server.waitLine(t, ctx, "PY_SERVER_OWNER_ZERO")
server.waitExit(t, ctx)
}
func TestGoAttachedResourceTwoHopLifecycleAgainstTypeScriptServer(t *testing.T) {
// Start the TypeScript attached-server subprocess with a bounded context.
ctx, cancel := context.WithTimeout(t.Context(), crossRuntimeFixtureTimeout)
defer cancel()
server := startCrossRuntimeProcess(
t,
ctx,
"bun",
"run",
"./bldr/resource/testdata/cross-runtime-ts-core-server.ts",
"127.0.0.1:0",
"attached",
)
// Build the Go ResourceClient against the TypeScript server listener.
ready := server.waitLine(t, ctx, "READY ")
service := resource.NewSRPCResourceServiceClient(
newCrossRuntimeTCPClient(strings.TrimPrefix(ready, "READY ")),
)
client, err := resource_client.NewClient(ctx, service)
if err != nil {
t.Fatalf("start Go ResourceClient: %v", err)
}
// Prepare the attached child channels and the fake Engine invoker.
var childReleaseCount atomic.Int32
var childAbortCount atomic.Int32
childIDCh := make(chan uint32, 1)
blockActive := make(chan struct{}, 1)
childAborted := make(chan struct{}, 1)
childReleased := make(chan struct{}, 1)
attachedMux := srpc.InvokerFunc(func(serviceID, methodID string, strm srpc.Stream) (bool, error) {
// Handle only the attached Engine Construct method.
if serviceID != "test.AttachedEngine" || methodID != "Construct" {
return false, nil
}
// Receive and validate the construct request payload.
request := srpc.NewRawMessage(nil, true)
if err := strm.MsgRecv(request); err != nil {
return true, err
}
if string(request.GetData()) != "construct" {
return true, errors.New("attached construct request mismatch")
}
// Add the fake attached child resource under the requesting client owner.
owner, err := resource_server.MustGetResourceClientContext(strm.Context())
if err != nil {
return true, err
}
childID, err := owner.AddResource(srpc.InvokerFunc(func(serviceID, methodID string, strm srpc.Stream) (bool, error) {
// Handle only the attached child service.
if serviceID != "test.AttachedChild" {
return false, nil
}
// Receive the child request payload before dispatching the method.
request := srpc.NewRawMessage(nil, true)
if err := strm.MsgRecv(request); err != nil {
return true, err
}
// Dispatch the child Invoke and Block methods.
switch methodID {
case "Invoke":
if string(request.GetData()) != "invoke" {
return true, errors.New("attached child invoke request mismatch")
}
return true, strm.MsgSend(srpc.NewRawMessage([]byte("nested-success"), true))
case "Block":
if string(request.GetData()) != "block" {
return true, errors.New("attached child block request mismatch")
}
// Send the active frame and hold until the stream context aborts.
if err := strm.MsgSend(srpc.NewRawMessage([]byte("active"), true)); err != nil {
return true, err
}
blockActive <- struct{}{}
<-strm.Context().Done()
childAbortCount.Add(1)
childAborted <- struct{}{}
return true, context.Cause(strm.Context())
default:
return false, nil
}
}), func() {
childReleaseCount.Add(1)
childReleased <- struct{}{}
})
// Report the child ID to the test and answer the construct call.
if err != nil {
return true, err
}
childIDCh <- childID
return true, strm.MsgSend(srpc.NewRawMessage(encodeCrossRuntimeID(childID), true))
})
// Attach the fake Engine tree to the client and validate the returned root ID.
attachedRootID, err := client.AttachResourceTree(ctx, "fake-engine", attachedMux)
if err != nil {
t.Fatalf("attach fake Engine tree: %v\n%s", err, server.logs())
}
if attachedRootID == 0 {
t.Fatal("AddAck returned an empty attached root ID")
}
// Open the TypeScript root resource and construct the attached object through it.
root := client.AccessRootResource()
rootClient, err := root.GetClient()
if err != nil {
t.Fatalf("get TypeScript root client: %v", err)
}
objectData, err := crossRuntimeCall(
ctx,
rootClient,
"test.Root",
"ConstructAttachedObject",
encodeCrossRuntimeID(attachedRootID),
)
if err != nil {
t.Fatalf("construct TypeScript ObjectType-like child: %v\n%s", err, server.logs())
}
// Match the TypeScript AddAck marker against the attached root ID.
if len(objectData) != 4 {
t.Fatalf("object ID length = %d, want 4", len(objectData))
}
objectID := binary.BigEndian.Uint32(objectData)
addAck := server.waitLine(t, ctx, "ATTACHED_ADD_ACK ")
if strings.TrimPrefix(addAck, "ATTACHED_ADD_ACK ") != strconv.FormatUint(uint64(attachedRootID), 10) {
t.Fatalf("TypeScript AddAck ID marker = %q, Go AddAck ID = %d", addAck, attachedRootID)
}
// Match the child-added marker against the ID reported by the fake Engine.
childMarker := server.waitLine(t, ctx, "ATTACHED_CHILD_ADDED ")
var childID uint32
select {
case childID = <-childIDCh:
case <-ctx.Done():
t.Fatalf("timed out receiving attached child ID after %q", childMarker)
}
if strings.TrimPrefix(childMarker, "ATTACHED_CHILD_ADDED ") != strconv.FormatUint(uint64(childID), 10) {
t.Fatalf("TypeScript child ID marker = %q, Go child ID = %d", childMarker, childID)
}
// Use the constructed object and confirm the nested child handled the call.
object := client.CreateResourceReference(objectID)
objectClient, err := object.GetClient()
if err != nil {
t.Fatalf("get constructed object client: %v", err)
}
data, err := crossRuntimeCall(ctx, objectClient, "test.AttachedObject", "Use", []byte("use"))
if err != nil {
t.Fatalf("use nested attached Go child after construction: %v\n%s", err, server.logs())
}
if string(data) != "nested-success" {
t.Fatalf("nested response = %q, want nested-success", data)
}
// Verify the child stayed attached and the client generation stayed live.
server.waitLine(t, ctx, "ATTACHED_USE_COMPLETE")
if got := childReleaseCount.Load(); got != 0 {
t.Fatalf("attached child detached before response: release count = %d", got)
}
server.send(t, "CHECK_LIVE")
server.waitLine(t, ctx, "ATTACHED_LIVE_NO_CANCEL")
select {
case <-client.Done():
t.Fatal("ResourceClient generation retired after successful two-hop call")
default:
}
// Start the nested Block call and wait for the fake child to become active.
blockResult := make(chan error, 1)
go func() {
_, callErr := crossRuntimeCall(ctx, objectClient, "test.AttachedObject", "Block", []byte("block"))
blockResult <- callErr
}()
select {
case <-blockActive:
case <-ctx.Done():
t.Fatalf("nested attached method did not become active\n%s", server.logs())
}
// Detach the attached child and confirm both sides observed the abort.
server.send(t, "DETACH_ATTACHED_CHILD")
server.waitLine(t, ctx, "ATTACHED_CHILD_DETACHED")
select {
case <-childReleased:
case <-ctx.Done():
t.Fatal("nested attached child detach was not acknowledged")
}
server.waitLine(t, ctx, "ATTACHED_BLOCK_ABORTED")
select {
case <-childAborted:
case <-ctx.Done():
t.Fatal("nested attached Go handler did not observe detach")
}
// Verify the detached call failed exactly once and the child released once.
select {
case err := <-blockResult:
if err == nil {
t.Fatal("detached nested attached method returned success")
}
case <-ctx.Done():
t.Fatal("detached nested attached method did not return")
}
if got := childAbortCount.Load(); got != 1 {
t.Fatalf("attached child abort count = %d, want 1", got)
}
if got := childReleaseCount.Load(); got != 1 {
t.Fatalf("attached child release count = %d, want 1", got)
}
server.send(t, "CHECK_ABORT_ONCE")
server.waitLine(t, ctx, "ATTACHED_ABORT_ONCE")
// Release the object, root, and client, then confirm the generation ends.
object.Release()
root.Release()
client.Release()
select {
case <-client.Done():
case <-ctx.Done():
t.Fatal("timed out ending Go ResourceClient generation")
}
if got := childReleaseCount.Load(); got != 1 {
t.Fatalf("attached child release count after client Done = %d, want 1", got)
}
// Verify the TypeScript server drains to owner zero and exits cleanly.
server.send(t, "CHECK_CLEAN")
server.waitLine(t, ctx, "TS_SERVER_OWNER_ZERO")
if got := childReleaseCount.Load(); got != 1 {
t.Fatalf("attached child release count after TypeScript owner zero = %d, want 1", got)
}
server.waitExit(t, ctx)
if got := childReleaseCount.Load(); got != 1 {
t.Fatalf("attached child release count after server exit = %d, want 1", got)
}
}
func encodeCrossRuntimeID(id uint32) []byte {
data := make([]byte, 4)
binary.BigEndian.PutUint32(data, id)
return data
}
func newCrossRuntimeTCPClient(address string) srpc.Client {
return srpc.NewClient(func(
ctx context.Context,
handler srpc.PacketDataHandler,
closeHandler srpc.CloseHandler,
) (srpc.PacketWriter, error) {
// Dial the cross-runtime server and start its packet read pump.
connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", address)
if err != nil {
return nil, err
}
packets := srpc.NewPacketReadWriter(connection)
go packets.ReadPump(handler, closeHandler)
return packets, nil
})
}
func crossRuntimeCall(
ctx context.Context,
client srpc.Client,
service string,
method string,
data []byte,
) ([]byte, error) {
response := srpc.NewRawMessage(nil, true)
err := client.ExecCall(
ctx,
service,
method,
srpc.NewRawMessage(data, true),
response,
)
return response.GetData(), err
}