170 lines
8.2 KiB
Go
170 lines
8.2 KiB
Go
package agent
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
|
|
"github.com/rvbox/rvbox/internal/agentproto"
|
|
"github.com/rvbox/rvbox/internal/client/spool"
|
|
"github.com/rvbox/rvbox/internal/domain"
|
|
"google.golang.org/protobuf/proto"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
func TestRunOnceReconcilesReplaysAndAdvertisesCapacity_HP_RUNTIME_01(t *testing.T) {
|
|
ctx := context.Background()
|
|
store, err := spool.Open(ctx, spool.Options{DataDir: filepath.Join(t.TempDir(), "spool"), BusyTimeout: time.Second})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = store.Close() })
|
|
issue, err := domain.ParseUUIDv7("019c46f1-1d02-7000-8000-0000000000b1")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
hash := [32]byte{1}
|
|
if _, err := store.AcceptCommand(ctx, spool.Command{IssueUUID: issue, ImmutableSHA256: hash, Revision: 1, Phase: uint32(rvboxv1.CommandLifecycle_COMMAND_ACCEPTED), ExecutionSpec: []byte("spec")}, time.Now().UTC()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
eventPayload, err := proto.Marshal(&rvboxv1.CommandEvent{Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := store.AppendEvent(ctx, issue, spool.EventInput{Kind: 4, Compression: 1, RawBytes: uint64(len(eventPayload)), Payload: eventPayload, CreatedAt: time.Now().UTC()}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := store.AssignSendWindow(ctx, issue, 1, 1<<20); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
welcome, _ := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_ServerWelcome{ServerWelcome: &rvboxv1.ServerWelcome{SelectedProtocol: &rvboxv1.ProtocolVersion{Major: 1, Minor: 0}, ServerTime: timestamppb.Now()}}})
|
|
reconcileRequest, _ := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_ReconcileRequest{ReconcileRequest: &rvboxv1.ReconcileRequest{}}})
|
|
reconcileResult, _ := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_ReconcileResult{ReconcileResult: &rvboxv1.ReconcileResult{}}})
|
|
transport := &runnerTransport{reads: [][]byte{welcome, reconcileRequest, reconcileResult}, terminal: errors.New("transport closed")}
|
|
hello := &rvboxv1.ClientHello{ClientId: "runner-client", SupportedProtocol: &rvboxv1.ProtocolRange{Major: 1, MinMinor: 0, MaxMinor: 0}, DaemonVersion: "test", Platform: rvboxv1.Platform_PLATFORM_WINDOWS, Architecture: "amd64", DaemonCwd: `C:\`, SupportedShells: []rvboxv1.ShellType{rvboxv1.ShellType_SHELL_POWERSHELL}, ClientInstanceId: store.ClientInstanceID().String(), MaxRunningCommands: 1, MaxQueuedCommands: 1, SentAt: timestamppb.Now()}
|
|
err = RunOnce(ctx, RunnerOptions{Store: store, Dial: func(context.Context) (Transport, error) { return transport, nil }, Hello: hello, Limits: agentproto.DefaultLimits(), Backoff: BackoffOptions{Initial: time.Millisecond, Maximum: time.Millisecond, StableReset: time.Second}, Jitter: func(value time.Duration) time.Duration { return 0 }, Now: func() time.Time { return time.Now().UTC() }})
|
|
if !errors.Is(err, transport.terminal) {
|
|
t.Fatalf("RunOnce error = %v, want transport close", err)
|
|
}
|
|
transport.mu.Lock()
|
|
writes := append([][]byte(nil), transport.writes...)
|
|
transport.mu.Unlock()
|
|
if len(writes) != 4 {
|
|
t.Fatalf("wire write count = %d, want hello/snapshot/event/capacity", len(writes))
|
|
}
|
|
last, err := agentproto.DecodeEnvelope(writes[len(writes)-1], agentproto.DefaultLimits(), rvboxv1.Platform_PLATFORM_WINDOWS)
|
|
if err != nil || last.GetClientCapacity() == nil || last.GetClientCapacity().GetMaxRunningCommands() != 1 {
|
|
t.Fatalf("capacity advertisement = %#v, %v", last, err)
|
|
}
|
|
eventEnvelope, err := agentproto.DecodeEnvelope(writes[2], agentproto.DefaultLimits(), rvboxv1.Platform_PLATFORM_WINDOWS)
|
|
if err != nil || eventEnvelope.GetCommandEvent() == nil || eventEnvelope.GetCommandEvent().GetEventSeq() != 1 || eventEnvelope.GetCommandEvent().GetIssueUuid() != issue.String() {
|
|
t.Fatalf("replayed event = %#v, %v", eventEnvelope, err)
|
|
}
|
|
}
|
|
|
|
func TestRunnerOptionsRejectMissingJitter_HP_RUNTIME_02(t *testing.T) {
|
|
if err := (RunnerOptions{}).validate(); err == nil {
|
|
t.Fatal("empty runner options unexpectedly validated")
|
|
}
|
|
}
|
|
|
|
func TestRunReportsRetryableSessionError_HP_RUNTIME_04(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
store, err := spool.Open(ctx, spool.Options{DataDir: filepath.Join(t.TempDir(), "spool"), BusyTimeout: time.Second})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = store.Close() })
|
|
want := errors.New("dial refused")
|
|
reported := make(chan error, 1)
|
|
hello := &rvboxv1.ClientHello{ClientId: "runner-client", SupportedProtocol: &rvboxv1.ProtocolRange{Major: 1, MinMinor: 0, MaxMinor: 0}, DaemonVersion: "test", Platform: rvboxv1.Platform_PLATFORM_WINDOWS, Architecture: "amd64", DaemonCwd: `C:\\`, SupportedShells: []rvboxv1.ShellType{rvboxv1.ShellType_SHELL_POWERSHELL}, ClientInstanceId: store.ClientInstanceID().String(), MaxRunningCommands: 1, MaxQueuedCommands: 1, SentAt: timestamppb.Now()}
|
|
err = Run(ctx, RunnerOptions{
|
|
Store: store, Dial: func(context.Context) (Transport, error) { return nil, want }, Hello: hello,
|
|
Limits: agentproto.DefaultLimits(), Backoff: BackoffOptions{Initial: time.Millisecond, Maximum: time.Millisecond, StableReset: time.Second},
|
|
Jitter: func(time.Duration) time.Duration { return 0 }, Now: func() time.Time { return time.Now().UTC() },
|
|
OnSessionError: func(got error) { reported <- got; cancel() },
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("Run error = %v, want graceful cancellation", err)
|
|
}
|
|
select {
|
|
case got := <-reported:
|
|
if !errors.Is(got, want) {
|
|
t.Fatalf("reported error = %v, want %v", got, want)
|
|
}
|
|
default:
|
|
t.Fatal("retryable session error was not reported")
|
|
}
|
|
}
|
|
|
|
func TestFlushEventsAssignsAndSendsOnlyUnacknowledgedRows_HP_RUNTIME_03(t *testing.T) {
|
|
ctx := context.Background()
|
|
store, err := spool.Open(ctx, spool.Options{DataDir: filepath.Join(t.TempDir(), "spool"), BusyTimeout: time.Second})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer store.Close()
|
|
issue, err := domain.ParseUUIDv7("019c46f1-1d02-7000-8000-0000000000b2")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
hash := [32]byte{2}
|
|
if _, err := store.AcceptCommand(ctx, spool.Command{IssueUUID: issue, ImmutableSHA256: hash, Revision: 1, Phase: uint32(rvboxv1.CommandLifecycle_COMMAND_ACCEPTED)}, time.Now().UTC()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
payload, err := proto.Marshal(&rvboxv1.CommandEvent{Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := store.AppendEvent(ctx, issue, spool.EventInput{Kind: 4, Compression: 1, RawBytes: uint64(len(payload)), Payload: payload, CreatedAt: time.Now().UTC()}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
transport := &runnerTransport{}
|
|
sent := map[domain.UUID]uint64{}
|
|
session := Session{ID: "session", Generation: 1}
|
|
if err := flushEvents(ctx, transport, store, session, issue, agentproto.DefaultLimits(), sent); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if sent[issue] != 1 || len(transport.writes) != 1 {
|
|
t.Fatalf("sent cursor/writes = %d/%d", sent[issue], len(transport.writes))
|
|
}
|
|
if err := flushEvents(ctx, transport, store, session, issue, agentproto.DefaultLimits(), sent); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(transport.writes) != 1 {
|
|
t.Fatalf("already-sent event was duplicated: %d writes", len(transport.writes))
|
|
}
|
|
}
|
|
|
|
type runnerTransport struct {
|
|
mu sync.Mutex
|
|
reads [][]byte
|
|
writes [][]byte
|
|
terminal error
|
|
}
|
|
|
|
func (transport *runnerTransport) Write(_ context.Context, payload []byte) error {
|
|
transport.mu.Lock()
|
|
defer transport.mu.Unlock()
|
|
transport.writes = append(transport.writes, append([]byte(nil), payload...))
|
|
return nil
|
|
}
|
|
|
|
func (transport *runnerTransport) Read(context.Context) ([]byte, error) {
|
|
transport.mu.Lock()
|
|
defer transport.mu.Unlock()
|
|
if len(transport.reads) == 0 {
|
|
return nil, transport.terminal
|
|
}
|
|
payload := transport.reads[0]
|
|
transport.reads = transport.reads[1:]
|
|
return append([]byte(nil), payload...), nil
|
|
}
|
|
|
|
func (transport *runnerTransport) Close() error { return nil }
|