555 lines
13 KiB
Go
555 lines
13 KiB
Go
package localcontrol
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
|
|
"iop/packages/go/agentstate"
|
|
iop "iop/proto/gen/iop"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type recordingHost struct {
|
|
mu sync.Mutex
|
|
readCalls map[string]int
|
|
mutationCalls map[string]int
|
|
lastCommand ProjectCommand
|
|
readError error
|
|
mutationError error
|
|
}
|
|
|
|
func newRecordingHost() *recordingHost {
|
|
return &recordingHost{
|
|
readCalls: make(map[string]int),
|
|
mutationCalls: make(map[string]int),
|
|
}
|
|
}
|
|
|
|
type blockingHost struct {
|
|
*recordingHost
|
|
entered chan struct{}
|
|
release <-chan struct{}
|
|
|
|
countsMu sync.Mutex
|
|
invocations int
|
|
effectiveCommands map[string]struct{}
|
|
}
|
|
|
|
func newBlockingHost(release <-chan struct{}) *blockingHost {
|
|
return &blockingHost{
|
|
recordingHost: newRecordingHost(),
|
|
entered: make(chan struct{}, 1),
|
|
release: release,
|
|
effectiveCommands: make(map[string]struct{}),
|
|
}
|
|
}
|
|
|
|
func (h *blockingHost) StartProject(
|
|
_ context.Context,
|
|
command ProjectCommand,
|
|
) (MutationResult, error) {
|
|
h.recordingHost.mu.Lock()
|
|
h.mutationCalls[OperationProjectStart]++
|
|
h.lastCommand = command
|
|
h.recordingHost.mu.Unlock()
|
|
|
|
h.countsMu.Lock()
|
|
h.invocations++
|
|
if _, exists := h.effectiveCommands[command.CommandID]; !exists {
|
|
h.effectiveCommands[command.CommandID] = struct{}{}
|
|
}
|
|
h.countsMu.Unlock()
|
|
|
|
select {
|
|
case h.entered <- struct{}{}:
|
|
default:
|
|
}
|
|
<-h.release
|
|
return MutationResult{
|
|
SubjectID: command.ProjectID,
|
|
State: OperationProjectStart,
|
|
Summary: "project command applied",
|
|
}, nil
|
|
}
|
|
|
|
func (h *blockingHost) mutationStats() (int, int) {
|
|
h.countsMu.Lock()
|
|
defer h.countsMu.Unlock()
|
|
return h.invocations, len(h.effectiveCommands)
|
|
}
|
|
|
|
func (h *recordingHost) recordRead(operation, subject string) (StatusSnapshot, error) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
h.readCalls[operation]++
|
|
if h.readError != nil {
|
|
return StatusSnapshot{}, h.readError
|
|
}
|
|
return StatusSnapshot{
|
|
SubjectID: subject,
|
|
State: "ready",
|
|
Summary: "state available",
|
|
Entries: []StatusEntry{{
|
|
Kind: operation,
|
|
SubjectID: subject,
|
|
State: "ready",
|
|
Summary: "entry available",
|
|
}},
|
|
}, nil
|
|
}
|
|
|
|
func (h *recordingHost) RuntimeStatus(context.Context) (StatusSnapshot, error) {
|
|
return h.recordRead(OperationRuntimeStatus, "runtime")
|
|
}
|
|
|
|
func (h *recordingHost) ProjectStatus(
|
|
_ context.Context,
|
|
projectID string,
|
|
) (StatusSnapshot, error) {
|
|
return h.recordRead(OperationProjectStatus, projectID)
|
|
}
|
|
|
|
func (h *recordingHost) OverlayStatus(
|
|
_ context.Context,
|
|
projectID string,
|
|
workUnitID string,
|
|
) (StatusSnapshot, error) {
|
|
return h.recordRead(OperationOverlayStatus, projectID+":"+workUnitID)
|
|
}
|
|
|
|
func (h *recordingHost) IntegrationStatus(
|
|
_ context.Context,
|
|
projectID string,
|
|
workUnitID string,
|
|
) (StatusSnapshot, error) {
|
|
return h.recordRead(OperationIntegrationStatus, projectID+":"+workUnitID)
|
|
}
|
|
|
|
func (h *recordingHost) BlockerList(
|
|
_ context.Context,
|
|
projectID string,
|
|
) (StatusSnapshot, error) {
|
|
return h.recordRead(OperationBlockerList, projectID)
|
|
}
|
|
|
|
func (h *recordingHost) ProcessStatus(
|
|
_ context.Context,
|
|
clientKind string,
|
|
) (StatusSnapshot, error) {
|
|
return h.recordRead(OperationProcessStatus, clientKind)
|
|
}
|
|
|
|
func (h *recordingHost) mutate(
|
|
operation string,
|
|
command ProjectCommand,
|
|
) (MutationResult, error) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
h.mutationCalls[operation]++
|
|
h.lastCommand = command
|
|
if h.mutationError != nil {
|
|
return MutationResult{}, h.mutationError
|
|
}
|
|
return MutationResult{
|
|
SubjectID: command.ProjectID,
|
|
State: operation,
|
|
Summary: "project command applied",
|
|
}, nil
|
|
}
|
|
|
|
func (h *recordingHost) StartProject(
|
|
_ context.Context,
|
|
command ProjectCommand,
|
|
) (MutationResult, error) {
|
|
return h.mutate(OperationProjectStart, command)
|
|
}
|
|
|
|
func (h *recordingHost) StopProject(
|
|
_ context.Context,
|
|
command ProjectCommand,
|
|
) (MutationResult, error) {
|
|
return h.mutate(OperationProjectStop, command)
|
|
}
|
|
|
|
func (h *recordingHost) ResumeProject(
|
|
_ context.Context,
|
|
command ProjectCommand,
|
|
) (MutationResult, error) {
|
|
return h.mutate(OperationProjectResume, command)
|
|
}
|
|
|
|
func (h *recordingHost) callCount(operation string) int {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
return h.readCalls[operation] + h.mutationCalls[operation]
|
|
}
|
|
|
|
func (h *recordingHost) totalCalls() int {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
total := 0
|
|
for _, count := range h.readCalls {
|
|
total += count
|
|
}
|
|
for _, count := range h.mutationCalls {
|
|
total += count
|
|
}
|
|
return total
|
|
}
|
|
|
|
func newTestService(
|
|
t *testing.T,
|
|
statePath string,
|
|
daemonID string,
|
|
retention int,
|
|
host interface {
|
|
StateReader
|
|
ProjectController
|
|
},
|
|
) (*Service, *Ledger) {
|
|
t.Helper()
|
|
store, err := agentstate.NewStore(statePath)
|
|
if err != nil {
|
|
t.Fatalf("NewStore: %v", err)
|
|
}
|
|
ledger, err := NewLedger(context.Background(), store, daemonID, retention)
|
|
if err != nil {
|
|
t.Fatalf("NewLedger: %v", err)
|
|
}
|
|
service, err := NewService(host, host, ledger)
|
|
if err != nil {
|
|
t.Fatalf("NewService: %v", err)
|
|
}
|
|
return service, ledger
|
|
}
|
|
|
|
func TestServiceDispatchesEveryS11Operation(t *testing.T) {
|
|
t.Parallel()
|
|
host := newRecordingHost()
|
|
service, _ := newTestService(
|
|
t,
|
|
filepath.Join(t.TempDir(), "state.json"),
|
|
"daemon-service",
|
|
8,
|
|
host,
|
|
)
|
|
readRequests := map[string]*iop.AgentLocalEnvelope{
|
|
OperationRuntimeStatus: runtimeStatusRequest("read-runtime"),
|
|
OperationProjectStatus: readRequest(
|
|
"read-project",
|
|
OperationProjectStatus,
|
|
"project-a",
|
|
"",
|
|
"",
|
|
),
|
|
OperationOverlayStatus: readRequest(
|
|
"read-overlay",
|
|
OperationOverlayStatus,
|
|
"project-a",
|
|
"work-a",
|
|
"",
|
|
),
|
|
OperationIntegrationStatus: readRequest(
|
|
"read-integration",
|
|
OperationIntegrationStatus,
|
|
"project-a",
|
|
"work-a",
|
|
"",
|
|
),
|
|
OperationBlockerList: readRequest(
|
|
"read-blockers",
|
|
OperationBlockerList,
|
|
"project-a",
|
|
"",
|
|
"",
|
|
),
|
|
OperationProcessStatus: readRequest(
|
|
"read-process",
|
|
OperationProcessStatus,
|
|
"",
|
|
"",
|
|
"flutter",
|
|
),
|
|
}
|
|
for operation, request := range readRequests {
|
|
response := service.Handle(context.Background(), true, request)
|
|
if response.GetKind() != iop.AgentLocalKind_AGENT_LOCAL_KIND_RESPONSE ||
|
|
response.GetResponse().GetSnapshot() == nil {
|
|
t.Fatalf("%s response = %#v", operation, response)
|
|
}
|
|
if host.callCount(operation) != 1 {
|
|
t.Fatalf("%s calls = %d, want 1", operation, host.callCount(operation))
|
|
}
|
|
}
|
|
|
|
for index, operation := range []string{
|
|
OperationProjectStart,
|
|
OperationProjectStop,
|
|
OperationProjectResume,
|
|
} {
|
|
response := service.Handle(
|
|
context.Background(),
|
|
true,
|
|
projectRequest(
|
|
fmt.Sprintf("mutation-%d", index),
|
|
fmt.Sprintf("command-%d", index),
|
|
operation,
|
|
"project-a",
|
|
),
|
|
)
|
|
if response.GetKind() != iop.AgentLocalKind_AGENT_LOCAL_KIND_RESPONSE ||
|
|
!response.GetResponse().GetMutation().GetAccepted() {
|
|
t.Fatalf("%s response = %#v", operation, response)
|
|
}
|
|
if host.callCount(operation) != 1 {
|
|
t.Fatalf("%s calls = %d, want 1", operation, host.callCount(operation))
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestServiceRejectedFramesHaveZeroCalls(t *testing.T) {
|
|
t.Parallel()
|
|
host := newRecordingHost()
|
|
service, _ := newTestService(
|
|
t,
|
|
filepath.Join(t.TempDir(), "state.json"),
|
|
"daemon-rejected",
|
|
8,
|
|
host,
|
|
)
|
|
tests := []struct {
|
|
name string
|
|
authorized bool
|
|
request *iop.AgentLocalEnvelope
|
|
code string
|
|
}{
|
|
{
|
|
name: "unauthorized",
|
|
authorized: false,
|
|
request: projectRequest(
|
|
"denied-message",
|
|
"denied-command",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
),
|
|
code: ErrorPermissionDenied,
|
|
},
|
|
{
|
|
name: "malformed",
|
|
authorized: true,
|
|
request: runtimeStatusRequest(""),
|
|
code: ErrorMalformedFrame,
|
|
},
|
|
{
|
|
name: "unsupported client",
|
|
authorized: true,
|
|
request: clientRequest(
|
|
"client-message",
|
|
"client-command",
|
|
OperationClientStart,
|
|
),
|
|
code: ErrorUnsupportedOperation,
|
|
},
|
|
}
|
|
for _, test := range tests {
|
|
response := service.Handle(
|
|
context.Background(),
|
|
test.authorized,
|
|
test.request,
|
|
)
|
|
if response.GetError().GetCode() != test.code {
|
|
t.Fatalf("%s error = %q, want %q", test.name, response.GetError().GetCode(), test.code)
|
|
}
|
|
}
|
|
if calls := host.totalCalls(); calls != 0 {
|
|
t.Fatalf("rejected frame host calls = %d, want 0", calls)
|
|
}
|
|
}
|
|
|
|
func TestConcurrentIdenticalCommandConvergesOnCompletedResponse(t *testing.T) {
|
|
t.Parallel()
|
|
release := make(chan struct{})
|
|
host := newBlockingHost(release)
|
|
service, ledger := newTestService(
|
|
t,
|
|
filepath.Join(t.TempDir(), "state.json"),
|
|
"daemon-concurrent",
|
|
8,
|
|
host,
|
|
)
|
|
ownerRequest := projectRequest(
|
|
"message-owner",
|
|
"one-command",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
)
|
|
ownerResult := make(chan *iop.AgentLocalEnvelope, 1)
|
|
go func() {
|
|
ownerResult <- service.Handle(context.Background(), true, ownerRequest)
|
|
}()
|
|
<-host.entered
|
|
|
|
requestHash, err := immutableCommandHash(
|
|
ownerRequest.GetOperation(),
|
|
ownerRequest.GetRequest(),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("immutableCommandHash: %v", err)
|
|
}
|
|
begin, failure := ledger.BeginCommand(
|
|
context.Background(),
|
|
"one-command",
|
|
requestHash,
|
|
mutationResponse(
|
|
ownerRequest,
|
|
nil,
|
|
&iop.AgentLocalMutationResult{Accepted: true, State: "accepted"},
|
|
),
|
|
)
|
|
if failure != nil || begin.ownership != commandWaiter {
|
|
t.Fatalf("cancelled waiter begin = %#v, failure = %#v", begin, failure)
|
|
}
|
|
cancelled, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
if failure := ledger.WaitCommand(cancelled, begin.flight); failure == nil {
|
|
t.Fatal("cancelled waiter completed unexpectedly")
|
|
}
|
|
|
|
waiterResult := make(chan *iop.AgentLocalEnvelope, 1)
|
|
go func() {
|
|
waiterResult <- service.Handle(
|
|
context.Background(),
|
|
true,
|
|
projectRequest(
|
|
"message-waiter",
|
|
"one-command",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
),
|
|
)
|
|
}()
|
|
conflict := service.Handle(
|
|
context.Background(),
|
|
true,
|
|
projectRequest(
|
|
"message-conflict",
|
|
"one-command",
|
|
OperationProjectStart,
|
|
"project-b",
|
|
),
|
|
)
|
|
if conflict.GetError().GetCode() != ErrorCommandIDConflict {
|
|
t.Fatalf("in-flight conflict = %#v", conflict)
|
|
}
|
|
if invocations, effects := host.mutationStats(); invocations != 1 || effects != 1 {
|
|
t.Fatalf("before release invocations=%d effects=%d, want 1/1", invocations, effects)
|
|
}
|
|
|
|
close(release)
|
|
owner := <-ownerResult
|
|
waiter := <-waiterResult
|
|
if owner.GetResponse().GetMutation().GetState() == "accepted" {
|
|
t.Fatalf("owner received provisional response: %#v", owner)
|
|
}
|
|
if !proto.Equal(owner, waiter) {
|
|
t.Fatalf("waiter response differs from final response:\nowner=%v\nwaiter=%v", owner, waiter)
|
|
}
|
|
if owner.GetResponse().GetStateRevision() != 1 ||
|
|
owner.GetResponse().GetReplayCursor() != 1 {
|
|
t.Fatalf("completed response = %#v", owner)
|
|
}
|
|
if invocations, effects := host.mutationStats(); invocations != 1 || effects != 1 {
|
|
t.Fatalf("after release invocations=%d effects=%d, want 1/1", invocations, effects)
|
|
}
|
|
if view, failure := ledger.View(context.Background()); failure != nil ||
|
|
view.StateRevision != 1 || view.ReplayCursor != 1 {
|
|
t.Fatalf("ledger view = %#v, failure = %#v", view, failure)
|
|
}
|
|
events, _, failure := ledger.Replay(context.Background(), "daemon-concurrent", 0)
|
|
if failure != nil || len(events) != 1 || events[0].GetEventSequence() != 1 {
|
|
t.Fatalf("retained events = %#v, failure = %#v", events, failure)
|
|
}
|
|
}
|
|
|
|
func runtimeStatusRequest(messageID string) *iop.AgentLocalEnvelope {
|
|
return readRequest(messageID, OperationRuntimeStatus, "", "", "")
|
|
}
|
|
|
|
func readRequest(
|
|
messageID string,
|
|
operation string,
|
|
projectID string,
|
|
workUnitID string,
|
|
clientKind string,
|
|
) *iop.AgentLocalEnvelope {
|
|
return &iop.AgentLocalEnvelope{
|
|
ProtocolVersion: ProtocolVersion,
|
|
Kind: iop.AgentLocalKind_AGENT_LOCAL_KIND_REQUEST,
|
|
MessageId: messageID,
|
|
Operation: operation,
|
|
Payload: &iop.AgentLocalEnvelope_Request{
|
|
Request: &iop.AgentLocalRequest{
|
|
Payload: &iop.AgentLocalRequest_Read{
|
|
Read: &iop.AgentLocalReadRequest{
|
|
ProjectId: projectID,
|
|
WorkUnitId: workUnitID,
|
|
ClientKind: clientKind,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
func projectRequest(
|
|
messageID string,
|
|
commandID string,
|
|
operation string,
|
|
projectID string,
|
|
) *iop.AgentLocalEnvelope {
|
|
return &iop.AgentLocalEnvelope{
|
|
ProtocolVersion: ProtocolVersion,
|
|
Kind: iop.AgentLocalKind_AGENT_LOCAL_KIND_REQUEST,
|
|
MessageId: messageID,
|
|
Operation: operation,
|
|
Payload: &iop.AgentLocalEnvelope_Request{
|
|
Request: &iop.AgentLocalRequest{
|
|
CommandId: commandID,
|
|
Payload: &iop.AgentLocalRequest_Project{
|
|
Project: &iop.AgentLocalProjectRequest{
|
|
ProjectId: projectID,
|
|
WorkspaceId: "workspace-a",
|
|
MilestoneId: "milestone-a",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
func clientRequest(
|
|
messageID string,
|
|
commandID string,
|
|
operation string,
|
|
) *iop.AgentLocalEnvelope {
|
|
return &iop.AgentLocalEnvelope{
|
|
ProtocolVersion: ProtocolVersion,
|
|
Kind: iop.AgentLocalKind_AGENT_LOCAL_KIND_REQUEST,
|
|
MessageId: messageID,
|
|
Operation: operation,
|
|
Payload: &iop.AgentLocalEnvelope_Request{
|
|
Request: &iop.AgentLocalRequest{
|
|
CommandId: commandID,
|
|
Payload: &iop.AgentLocalRequest_Client{
|
|
Client: &iop.AgentLocalClientRequest{
|
|
ClientKind: "flutter",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|