353 lines
9.5 KiB
Go
353 lines
9.5 KiB
Go
package localcontrol
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"path/filepath"
|
|
"testing"
|
|
|
|
"iop/packages/go/agentstate"
|
|
iop "iop/proto/gen/iop"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
func TestCommandIdempotencySurvivesRestart(t *testing.T) {
|
|
t.Parallel()
|
|
statePath := filepath.Join(t.TempDir(), "state", "manager.json")
|
|
host := newRecordingHost()
|
|
firstService, _ := newTestService(
|
|
t,
|
|
statePath,
|
|
"daemon-restart",
|
|
8,
|
|
host,
|
|
)
|
|
request := projectRequest(
|
|
"message-first",
|
|
"command-stable",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
)
|
|
first := firstService.Handle(context.Background(), true, request)
|
|
if first.GetResponse().GetMutation() == nil {
|
|
t.Fatalf("first response = %#v", first)
|
|
}
|
|
|
|
restartedService, _ := newTestService(
|
|
t,
|
|
statePath,
|
|
"daemon-restart",
|
|
8,
|
|
host,
|
|
)
|
|
replayedRequest := cloneEnvelope(request)
|
|
replayedRequest.MessageId = "message-after-restart"
|
|
second := restartedService.Handle(
|
|
context.Background(),
|
|
true,
|
|
replayedRequest,
|
|
)
|
|
if !proto.Equal(first, second) {
|
|
t.Fatalf("restart replay changed response:\nfirst=%v\nsecond=%v", first, second)
|
|
}
|
|
if calls := host.callCount(OperationProjectStart); calls != 1 {
|
|
t.Fatalf("project.start calls = %d, want 1", calls)
|
|
}
|
|
}
|
|
|
|
func TestAcceptedButIncompleteCommandReconcilesAfterRestart(t *testing.T) {
|
|
t.Parallel()
|
|
statePath := filepath.Join(t.TempDir(), "state", "manager.json")
|
|
release := make(chan struct{})
|
|
close(release)
|
|
host := newBlockingHost(release)
|
|
store, err := agentstate.NewStore(statePath)
|
|
if err != nil {
|
|
t.Fatalf("NewStore: %v", err)
|
|
}
|
|
ledger, err := NewLedger(context.Background(), store, "daemon-recovery", 8)
|
|
if err != nil {
|
|
t.Fatalf("NewLedger: %v", err)
|
|
}
|
|
request := projectRequest(
|
|
"message-before-restart",
|
|
"command-recovery",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
)
|
|
requestHash, err := immutableCommandHash(request.GetOperation(), request.GetRequest())
|
|
if err != nil {
|
|
t.Fatalf("immutableCommandHash: %v", err)
|
|
}
|
|
begin, failure := ledger.BeginCommand(
|
|
context.Background(),
|
|
request.GetRequest().GetCommandId(),
|
|
requestHash,
|
|
mutationResponse(
|
|
request,
|
|
nil,
|
|
&iop.AgentLocalMutationResult{Accepted: true, State: "accepted"},
|
|
),
|
|
)
|
|
if failure != nil || begin.ownership != commandOwner {
|
|
t.Fatalf("initial acceptance = %#v, failure = %#v", begin, failure)
|
|
}
|
|
_, err = host.StartProject(context.Background(), ProjectCommand{
|
|
CommandID: "command-recovery",
|
|
ProjectID: "project-a",
|
|
WorkspaceID: "workspace-a",
|
|
MilestoneID: "milestone-a",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("initial controller invocation: %v", err)
|
|
}
|
|
ledger.ReleaseCommandOwner("command-recovery", begin.flight)
|
|
|
|
restartedService, restartedLedger := newTestService(
|
|
t,
|
|
statePath,
|
|
"daemon-recovery",
|
|
8,
|
|
host,
|
|
)
|
|
conflict := restartedService.Handle(
|
|
context.Background(),
|
|
true,
|
|
projectRequest(
|
|
"message-conflict",
|
|
"command-recovery",
|
|
OperationProjectStart,
|
|
"project-b",
|
|
),
|
|
)
|
|
if conflict.GetError().GetCode() != ErrorCommandIDConflict {
|
|
t.Fatalf("incomplete conflict = %#v", conflict)
|
|
}
|
|
|
|
recoveredRequest := cloneEnvelope(request)
|
|
recoveredRequest.MessageId = "message-after-restart"
|
|
completed := restartedService.Handle(context.Background(), true, recoveredRequest)
|
|
if completed.GetResponse().GetMutation().GetState() == "accepted" {
|
|
t.Fatalf("recovery returned provisional response: %#v", completed)
|
|
}
|
|
if invocations, effects := host.mutationStats(); invocations != 2 || effects != 1 {
|
|
t.Fatalf("recovery invocations=%d effects=%d, want 2/1", invocations, effects)
|
|
}
|
|
if host.lastCommand.CommandID != "command-recovery" {
|
|
t.Fatalf("recovery command_id = %q", host.lastCommand.CommandID)
|
|
}
|
|
if completed.GetResponse().GetStateRevision() != 1 ||
|
|
completed.GetResponse().GetReplayCursor() != 1 {
|
|
t.Fatalf("completed response = %#v", completed)
|
|
}
|
|
if view, failure := restartedLedger.View(context.Background()); failure != nil ||
|
|
view.StateRevision != 1 || view.ReplayCursor != 1 {
|
|
t.Fatalf("restarted ledger view = %#v, failure = %#v", view, failure)
|
|
}
|
|
events, _, failure := restartedLedger.Replay(context.Background(), "daemon-recovery", 0)
|
|
if failure != nil || len(events) != 1 || events[0].GetEventSequence() != 1 {
|
|
t.Fatalf("restarted events = %#v, failure = %#v", events, failure)
|
|
}
|
|
replayedRequest := cloneEnvelope(recoveredRequest)
|
|
replayedRequest.MessageId = "message-replayed"
|
|
replayed := restartedService.Handle(context.Background(), true, replayedRequest)
|
|
if !proto.Equal(completed, replayed) {
|
|
t.Fatalf("completed replay changed response:\ncompleted=%v\nreplayed=%v", completed, replayed)
|
|
}
|
|
}
|
|
|
|
func TestCommandIDConflictHasZeroMutation(t *testing.T) {
|
|
t.Parallel()
|
|
host := newRecordingHost()
|
|
service, _ := newTestService(
|
|
t,
|
|
filepath.Join(t.TempDir(), "state.json"),
|
|
"daemon-conflict",
|
|
8,
|
|
host,
|
|
)
|
|
first := projectRequest(
|
|
"message-first",
|
|
"command-conflict",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
)
|
|
if response := service.Handle(context.Background(), true, first); response.GetError() != nil {
|
|
t.Fatalf("first response = %#v", response)
|
|
}
|
|
conflict := projectRequest(
|
|
"message-conflict",
|
|
"command-conflict",
|
|
OperationProjectStart,
|
|
"project-b",
|
|
)
|
|
response := service.Handle(context.Background(), true, conflict)
|
|
if response.GetError().GetCode() != ErrorCommandIDConflict {
|
|
t.Fatalf("conflict code = %q", response.GetError().GetCode())
|
|
}
|
|
if calls := host.callCount(OperationProjectStart); calls != 1 {
|
|
t.Fatalf("project.start calls = %d, want 1", calls)
|
|
}
|
|
}
|
|
|
|
func TestReplayGapRequiresSnapshot(t *testing.T) {
|
|
t.Parallel()
|
|
host := newRecordingHost()
|
|
service, _ := newTestService(
|
|
t,
|
|
filepath.Join(t.TempDir(), "state.json"),
|
|
"daemon-replay",
|
|
2,
|
|
host,
|
|
)
|
|
for index := 1; index <= 3; index++ {
|
|
response := service.Handle(
|
|
context.Background(),
|
|
true,
|
|
projectRequest(
|
|
fmt.Sprintf("message-%d", index),
|
|
fmt.Sprintf("command-%d", index),
|
|
OperationProjectStart,
|
|
"project-a",
|
|
),
|
|
)
|
|
if response.GetError() != nil {
|
|
t.Fatalf("mutation %d: %#v", index, response)
|
|
}
|
|
}
|
|
|
|
gap := runtimeStatusRequest("message-gap")
|
|
cursor := uint64(0)
|
|
gap.GetRequest().ReplayAfterSequence = &cursor
|
|
gap.GetRequest().ReplayDaemonId = "daemon-replay"
|
|
gapResponse := service.Handle(context.Background(), true, gap)
|
|
if gapResponse.GetError().GetCode() != ErrorReplayUnavailable ||
|
|
!gapResponse.GetError().GetSnapshotRequired() ||
|
|
gapResponse.GetError().GetReplayFloor() != 1 {
|
|
t.Fatalf("gap response = %#v", gapResponse)
|
|
}
|
|
|
|
snapshot := service.Handle(
|
|
context.Background(),
|
|
true,
|
|
runtimeStatusRequest("message-snapshot"),
|
|
)
|
|
if snapshot.GetResponse().GetSnapshot() == nil ||
|
|
snapshot.GetResponse().GetReplayCursor() != 3 {
|
|
t.Fatalf("snapshot response = %#v", snapshot)
|
|
}
|
|
|
|
replay := runtimeStatusRequest("message-replay")
|
|
cursor = 1
|
|
replay.GetRequest().ReplayAfterSequence = &cursor
|
|
replay.GetRequest().ReplayDaemonId = "daemon-replay"
|
|
replayResponse := service.Handle(context.Background(), true, replay)
|
|
events := replayResponse.GetResponse().GetReplayEvents()
|
|
if len(events) != 2 ||
|
|
events[0].GetEventSequence() != 2 ||
|
|
events[1].GetEventSequence() != 3 {
|
|
t.Fatalf("replayed events = %#v", events)
|
|
}
|
|
}
|
|
|
|
func TestReplayRejectsAnotherDaemonIdentity(t *testing.T) {
|
|
t.Parallel()
|
|
host := newRecordingHost()
|
|
service, _ := newTestService(
|
|
t,
|
|
filepath.Join(t.TempDir(), "state.json"),
|
|
"daemon-owned",
|
|
8,
|
|
host,
|
|
)
|
|
request := runtimeStatusRequest("message-foreign")
|
|
cursor := uint64(0)
|
|
request.GetRequest().ReplayAfterSequence = &cursor
|
|
request.GetRequest().ReplayDaemonId = "daemon-foreign"
|
|
response := service.Handle(context.Background(), true, request)
|
|
if response.GetError().GetCode() != ErrorReplayUnavailable ||
|
|
!response.GetError().GetSnapshotRequired() {
|
|
t.Fatalf("response = %#v", response)
|
|
}
|
|
}
|
|
|
|
func TestLedgerRejectsDaemonIdentityDrift(t *testing.T) {
|
|
t.Parallel()
|
|
statePath := filepath.Join(t.TempDir(), "state.json")
|
|
host := newRecordingHost()
|
|
service, _ := newTestService(
|
|
t,
|
|
statePath,
|
|
"daemon-original",
|
|
8,
|
|
host,
|
|
)
|
|
response := service.Handle(
|
|
context.Background(),
|
|
true,
|
|
projectRequest(
|
|
"message-original",
|
|
"command-original",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
),
|
|
)
|
|
if response.GetKind() != iop.AgentLocalKind_AGENT_LOCAL_KIND_RESPONSE {
|
|
t.Fatalf("response = %#v", response)
|
|
}
|
|
|
|
store, err := agentstate.NewStore(statePath)
|
|
if err != nil {
|
|
t.Fatalf("NewStore: %v", err)
|
|
}
|
|
if _, err := NewLedger(
|
|
context.Background(),
|
|
store,
|
|
"daemon-replacement",
|
|
8,
|
|
); err == nil {
|
|
t.Fatal("NewLedger accepted a different daemon identity")
|
|
}
|
|
}
|
|
|
|
func TestConcurrentSubscribersReceiveCommittedEvent(t *testing.T) {
|
|
t.Parallel()
|
|
host := newRecordingHost()
|
|
service, ledger := newTestService(
|
|
t,
|
|
filepath.Join(t.TempDir(), "state.json"),
|
|
"daemon-subscribers",
|
|
8,
|
|
host,
|
|
)
|
|
first, cancelFirst := ledger.Subscribe()
|
|
defer cancelFirst()
|
|
second, cancelSecond := ledger.Subscribe()
|
|
defer cancelSecond()
|
|
|
|
response := service.Handle(
|
|
context.Background(),
|
|
true,
|
|
projectRequest(
|
|
"message-subscribers",
|
|
"command-subscribers",
|
|
OperationProjectStart,
|
|
"project-a",
|
|
),
|
|
)
|
|
if response.GetKind() != iop.AgentLocalKind_AGENT_LOCAL_KIND_RESPONSE {
|
|
t.Fatalf("response = %#v", response)
|
|
}
|
|
for index, subscriber := range []<-chan *iop.AgentLocalEnvelope{first, second} {
|
|
select {
|
|
case event := <-subscriber:
|
|
if event.GetEvent().GetEventSequence() != 1 ||
|
|
event.GetEvent().GetEventType() != OperationProjectStart {
|
|
t.Fatalf("subscriber %d event = %#v", index, event)
|
|
}
|
|
default:
|
|
t.Fatalf("subscriber %d did not receive committed event", index)
|
|
}
|
|
}
|
|
}
|