A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
25 kB · 581 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582//nolint:goconstpackage controller
import ( "context" "encoding/json" "errors" "net/http" "net/http/httptest" "os" "path/filepath" "testing" "time"
"connectrpc.com/connect" agentv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/agent/v1" "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/agent/v1/agentv1connect" "forgejo.alexma.top/alexma233/composia/internal/core/config" "forgejo.alexma.top/alexma233/composia/internal/core/task" "forgejo.alexma.top/alexma233/composia/internal/platform/rpcutil" "forgejo.alexma.top/alexma233/composia/internal/platform/store" "google.golang.org/protobuf/types/known/timestamppb")
func TestAgentPullAndReportTaskFlow(t *testing.T) { t.Parallel()
db := openControllerTestDB(t) defer func() { _ = db.Close() }()
ctx := context.Background() repoDir := t.TempDir() createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./demo.caddy\n", "edge/composia-meta.yaml": "name: edge\nnodes:\n - main\ninfra:\n caddy:\n compose_service: caddy\n config_dir: /etc/caddy\n", }) logDir := filepath.Join(t.TempDir(), "logs") if err := os.MkdirAll(filepath.Join(logDir, "tasks"), 0o750); err != nil { t.Fatalf("create task log dir: %v", err) } if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 16, 59, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "demo", "edge"); err != nil { t.Fatalf("sync declared services: %v", err) } revision := currentRevision(t, repoDir) paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo"}) if err != nil { t.Fatalf("marshal deploy task params: %v", err) } logPath := filepath.Join(t.TempDir(), "logs", "task.log") if _, err := db.CreateTask(ctx, task.Record{ TaskID: "task-remote", Type: task.TypeDeploy, Source: task.SourceWeb, ServiceName: "demo", NodeID: "main", CreatedAt: time.Date(2026, 4, 4, 17, 0, 0, 0, time.UTC), ParamsJSON: string(paramsJSON), RepoRevision: revision, LogPath: logPath, }); err != nil { t.Fatalf("create task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil })
mux := http.NewServeMux() reportServer := &agentReportServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}}, availableNodeIDs: map[string]struct{}{"main": {}}, logState: &taskLogAckState{confirmedBy: make(map[string]uint64)}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()} reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(reportServer, connect.WithInterceptors(interceptor)) mux.Handle(reportPath, reportHandler) taskPath, taskHandler := agentv1connect.NewAgentTaskServiceHandler(&agentTaskServer{db: db}, connect.WithInterceptors(interceptor)) mux.Handle(taskPath, taskHandler) httpServer := httptest.NewUnstartedServer(mux) httpServer.EnableHTTP2 = true httpServer.StartTLS() defer httpServer.Close()
taskClient := agentv1connect.NewAgentTaskServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) reportClient := agentv1connect.NewAgentReportServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token")))
pulled, err := taskClient.PullNextTask(ctx, connect.NewRequest(&agentv1.PullNextTaskRequest{NodeId: "main", ExecutionProtocol: 1})) if err != nil { t.Fatalf("pull next task: %v", err) } if !pulled.Msg.GetHasTask() || pulled.Msg.GetTask().GetTaskId() != "task-remote" { t.Fatalf("unexpected pulled task: %+v", pulled.Msg) } if pulled.Msg.GetTask().GetServiceDir() == "" { t.Fatalf("expected pulled task to include service_dir") } executionID := pulled.Msg.GetTask().GetExecutionId() if _, err := taskClient.AcknowledgeTask(ctx, connect.NewRequest(&agentv1.AcknowledgeTaskRequest{TaskId: "task-remote", ExecutionId: executionID})); err != nil { t.Fatalf("acknowledge task: %v", err) }
startedAt := timestamppb.New(time.Date(2026, 4, 4, 17, 1, 0, 0, time.UTC)) finishedAt := timestamppb.New(time.Date(2026, 4, 4, 17, 2, 0, 0, time.UTC)) if _, err := reportClient.ReportTaskStepState(ctx, connect.NewRequest(&agentv1.ReportTaskStepStateRequest{TaskId: "task-remote", ExecutionId: executionID, StepName: agentv1.AgentTaskStepName_AGENT_TASK_STEP_NAME_RENDER, Status: agentv1.AgentTaskStatus_AGENT_TASK_STATUS_SUCCEEDED, StartedAt: startedAt, FinishedAt: finishedAt})); err != nil { t.Fatalf("report task step state: %v", err) } logStream := reportClient.UploadTaskLogs(ctx) if err := logStream.Send(&agentv1.UploadTaskLogsRequest{TaskId: "task-remote", ExecutionId: executionID, Seq: 1, SentAt: finishedAt, Content: "hello from agent\n"}); err != nil { t.Fatalf("send task logs: %v", err) } logAck, err := logStream.Receive() if err != nil { t.Fatalf("receive log ack: %v", err) } if logAck.GetLastConfirmedSeq() != 1 { t.Fatalf("expected last_confirmed_seq 1, got %d", logAck.GetLastConfirmedSeq()) } if err := logStream.CloseRequest(); err != nil { t.Fatalf("close log request: %v", err) } if err := logStream.CloseResponse(); err != nil { t.Fatalf("close log response: %v", err) } reportServer.logState = &taskLogAckState{confirmedBy: make(map[string]uint64)} replayStream := reportClient.UploadTaskLogs(ctx) if err := replayStream.Send(&agentv1.UploadTaskLogsRequest{TaskId: "task-remote", ExecutionId: executionID, Seq: 1, SentAt: finishedAt, Content: "hello from agent\n"}); err != nil { t.Fatalf("send replayed task logs: %v", err) } replayAck, err := replayStream.Receive() if err != nil { t.Fatalf("receive replay ack: %v", err) } if replayAck.GetLastConfirmedSeq() != 1 { t.Fatalf("expected replay ack seq 1, got %d", replayAck.GetLastConfirmedSeq()) } if err := replayStream.Send(&agentv1.UploadTaskLogsRequest{TaskId: "task-remote", ExecutionId: executionID, Seq: 2, SentAt: finishedAt, Content: "second line\n"}); err != nil { t.Fatalf("send second task log: %v", err) } secondAck, err := replayStream.Receive() if err != nil { t.Fatalf("receive second ack: %v", err) } if secondAck.GetLastConfirmedSeq() != 2 { t.Fatalf("expected second ack seq 2, got %d", secondAck.GetLastConfirmedSeq()) } if err := replayStream.CloseRequest(); err != nil { t.Fatalf("close replay request: %v", err) } if err := replayStream.CloseResponse(); err != nil { t.Fatalf("close replay response: %v", err) } if _, err := reportClient.ReportServiceInstanceStatus(ctx, connect.NewRequest(&agentv1.ReportServiceInstanceStatusRequest{ServiceName: "demo", NodeId: "main", TaskId: "task-remote", ExecutionId: executionID, RuntimeStatus: store.ServiceRuntimeRunning, ReportedAt: finishedAt})); err != nil { t.Fatalf("report service instance status: %v", err) } if _, err := reportClient.ReportTaskState(ctx, connect.NewRequest(&agentv1.ReportTaskStateRequest{TaskId: "task-remote", ExecutionId: executionID, Status: agentv1.AgentTaskStatus_AGENT_TASK_STATUS_SUCCEEDED, FinishedAt: finishedAt})); err != nil { t.Fatalf("report task state: %v", err) } event, err := db.NextTaskOutboxEvent(ctx, time.Now().UTC()) if err != nil { t.Fatalf("read task outbox: %v", err) } if err := reportServer.processTaskOutboxEvent(ctx, event); err != nil { t.Fatalf("process task outbox: %v", err) } if err := reportServer.processTaskOutboxEvent(ctx, event); err != nil { t.Fatalf("reprocess task outbox idempotently: %v", err) }
detail, err := db.GetTask(ctx, "task-remote") if err != nil { t.Fatalf("get task detail: %v", err) } if detail.Record.Status != task.StatusSucceeded { t.Fatalf("expected succeeded task, got %q", detail.Record.Status) } reloadTasks, totalCount, err := db.ListTasks(ctx, []string{string(task.StatusPending)}, []string{"edge"}, []string{"main"}, []string{string(task.TypeCaddyReload)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list caddy reload tasks: %v", err) } if totalCount != 1 || len(reloadTasks) != 1 { t.Fatalf("expected one queued caddy reload task, got total=%d tasks=%+v", totalCount, reloadTasks) } reloadDetail, err := db.GetTask(ctx, reloadTasks[0].TaskID) if err != nil { t.Fatalf("get caddy reload task: %v", err) } if reloadDetail.Record.Source != task.SourceWeb { t.Fatalf("expected caddy reload source web, got %q", reloadDetail.Record.Source) } syncTasks, syncCount, err := db.ListTasks(ctx, []string{string(task.StatusPending)}, []string{"edge"}, []string{"main"}, []string{string(task.TypeCaddySync)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list caddy sync tasks: %v", err) } if syncCount != 0 || len(syncTasks) != 0 { t.Fatalf("expected no separate caddy sync tasks, got total=%d tasks=%+v", syncCount, syncTasks) } if len(detail.Steps) != 1 || detail.Steps[0].StepName != task.StepRender { t.Fatalf("unexpected task steps: %+v", detail.Steps) } content, err := os.ReadFile(logPath) //nolint:gosec if err != nil { t.Fatalf("read task log: %v", err) } if string(content) != "hello from agent\nsecond line\n" { t.Fatalf("unexpected task log content %q", string(content)) } snapshot, err := db.GetServiceSnapshot(ctx, "demo") if err != nil { t.Fatalf("get service snapshot: %v", err) } if snapshot.RuntimeStatus != store.ServiceRuntimeRunning { t.Fatalf("expected runtime status running, got %q", snapshot.RuntimeStatus) }}
func TestReportServiceInstanceStatusRejectsMismatchedNode(t *testing.T) { t.Parallel()
db := openControllerTestDB(t) defer func() { _ = db.Close() }()
ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main", "other"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "demo"); err != nil { t.Fatalf("sync declared services: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil })
mux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(&agentReportServer{db: db}, connect.WithInterceptors(interceptor)) mux.Handle(reportPath, reportHandler) httpServer := httptest.NewUnstartedServer(mux) httpServer.EnableHTTP2 = true httpServer.StartTLS() defer httpServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token")))
_, err := reportClient.ReportServiceInstanceStatus(ctx, connect.NewRequest(&agentv1.ReportServiceInstanceStatusRequest{ ServiceName: "demo", NodeId: "other", RuntimeStatus: store.ServiceRuntimeRunning, ReportedAt: timestamppb.New(time.Date(2026, 4, 4, 17, 2, 0, 0, time.UTC)), })) if err == nil { t.Fatal("expected unauthenticated error") } var connectErr *connect.Error if !errors.As(err, &connectErr) { t.Fatalf("expected connect error, got %T", err) } if connectErr.Code() != connect.CodeUnauthenticated { t.Fatalf("expected unauthenticated, got %v", connectErr.Code()) } if got := connectErr.Message(); got != "node_id does not match bearer token" { t.Fatalf("unexpected error message %q", got) }}
func TestAgentPullNextTaskLongPollWaitsForNewTask(t *testing.T) { t.Parallel()
db := openControllerTestDB(t) defer func() { _ = db.Close() }()
ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "demo"); err != nil { t.Fatalf("sync declared services: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) notifier := newTaskQueueNotifier()
mux := http.NewServeMux() taskPath, taskHandler := agentv1connect.NewAgentTaskServiceHandler(&agentTaskServer{db: db, taskQueue: notifier, maxWait: 700 * time.Millisecond, retryInterval: 500 * time.Millisecond}, connect.WithInterceptors(interceptor)) mux.Handle(taskPath, taskHandler) httpServer := httptest.NewUnstartedServer(mux) httpServer.EnableHTTP2 = true httpServer.StartTLS() defer httpServer.Close()
taskClient := agentv1connect.NewAgentTaskServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) pullCtx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel()
responseCh := make(chan *agentv1.PullNextTaskResponse, 1) errCh := make(chan error, 1) go func() { response, err := taskClient.PullNextTask(pullCtx, connect.NewRequest(&agentv1.PullNextTaskRequest{NodeId: "main", ExecutionProtocol: 1})) if err != nil { errCh <- err return } responseCh <- response.Msg }()
waitForTaskQueueSubscriber(t, notifier) paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo"}) if err != nil { t.Fatalf("marshal deploy task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{ TaskID: "task-long-poll", Type: task.TypeDeploy, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", ParamsJSON: string(paramsJSON), RepoRevision: "deadbeef", CreatedAt: time.Date(2026, 4, 5, 10, 0, 0, 0, time.UTC), }); err != nil { t.Fatalf("create task: %v", err) } notifier.Notify()
select { case err := <-errCh: t.Fatalf("pull next task: %v", err) case response := <-responseCh: if !response.GetHasTask() || response.GetTask().GetTaskId() != "task-long-poll" { t.Fatalf("unexpected long-poll response: %+v", response) } case <-time.After(900 * time.Millisecond): t.Fatalf("timed out waiting for long-poll response") }}
func TestAgentPullNextTaskDoesNotWaitForAnotherNodeTask(t *testing.T) { t.Parallel()
db := openControllerTestDB(t) defer func() { _ = db.Close() }()
ctx := context.Background() if err := db.SyncConfiguredNodes(ctx, []string{"main", "node-2"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := db.SyncDeclaredServices(ctx, map[string][]string{"alpha": {"main"}, "bravo": {"node-2"}}); err != nil { t.Fatalf("sync declared services: %v", err) } runningStartedAt := time.Date(2026, 4, 5, 11, 0, 10, 0, time.UTC) if _, err := db.CreateTask(ctx, task.Record{ TaskID: "task-running", Type: task.TypeDeploy, Source: task.SourceCLI, ServiceName: "alpha", NodeID: "main", Status: task.StatusRunning, CreatedAt: time.Date(2026, 4, 5, 11, 0, 0, 0, time.UTC), StartedAt: &runningStartedAt, }); err != nil { t.Fatalf("create running task: %v", err) } paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "bravo"}) if err != nil { t.Fatalf("marshal pending task params: %v", err) } if _, err := db.CreateTask(ctx, task.Record{ TaskID: "task-pending", Type: task.TypeDeploy, Source: task.SourceCLI, ServiceName: "bravo", NodeID: "node-2", ParamsJSON: string(paramsJSON), RepoRevision: "cafebabe", CreatedAt: time.Date(2026, 4, 5, 11, 1, 0, 0, time.UTC), }); err != nil { t.Fatalf("create pending task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "node-2-token" { //nolint:gosec return "", assertError("unexpected token") } return "node-2", nil }) notifier := newTaskQueueNotifier()
mux := http.NewServeMux() taskPath, taskHandler := agentv1connect.NewAgentTaskServiceHandler(&agentTaskServer{db: db, taskQueue: notifier, maxWait: time.Second, retryInterval: 700 * time.Millisecond}, connect.WithInterceptors(interceptor)) mux.Handle(taskPath, taskHandler) httpServer := httptest.NewUnstartedServer(mux) httpServer.EnableHTTP2 = true httpServer.StartTLS() defer httpServer.Close()
taskClient := agentv1connect.NewAgentTaskServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("node-2-token"))) pullCtx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond) defer cancel()
responseCh := make(chan *agentv1.PullNextTaskResponse, 1) errCh := make(chan error, 1) go func() { response, err := taskClient.PullNextTask(pullCtx, connect.NewRequest(&agentv1.PullNextTaskRequest{NodeId: "node-2", ExecutionProtocol: 1})) if err != nil { errCh <- err return } responseCh <- response.Msg }()
select { case err := <-errCh: t.Fatalf("pull next task: %v", err) case response := <-responseCh: if !response.GetHasTask() || response.GetTask().GetTaskId() != "task-pending" { t.Fatalf("unexpected response while another node task is running: %+v", response) } case <-time.After(time.Second): t.Fatalf("timed out waiting for pending task") }}
func waitForTaskQueueSubscriber(t *testing.T, notifier *taskQueueNotifier) { t.Helper() deadline := time.Now().Add(time.Second) for { notifier.mu.Lock() subscriberCount := len(notifier.subscribers) notifier.mu.Unlock() if subscriberCount > 0 { return } if time.Now().After(deadline) { t.Fatal("timed out waiting for task queue subscriber") } time.Sleep(5 * time.Millisecond) }}
func TestAgentReportTaskStateSkipsCaddyReloadWhenServiceDoesNotUseCaddy(t *testing.T) { t.Parallel()
db := openControllerTestDB(t) defer func() { _ = db.Close() }()
ctx := context.Background() repoDir := t.TempDir() createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\n", "edge/composia-meta.yaml": "name: edge\nnodes:\n - main\ninfra:\n caddy:\n compose_service: caddy\n config_dir: /etc/caddy\n", }) logDir := filepath.Join(t.TempDir(), "logs") if err := os.MkdirAll(filepath.Join(logDir, "tasks"), 0o750); err != nil { t.Fatalf("create task log dir: %v", err) } if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "demo", "edge"); err != nil { t.Fatalf("sync declared services: %v", err) } revision := currentRevision(t, repoDir) paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo"}) if err != nil { t.Fatalf("marshal params: %v", err) } logPath := filepath.Join(t.TempDir(), "logs", "task.log") if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-no-caddy", Type: task.TypeDeploy, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", ParamsJSON: string(paramsJSON), RepoRevision: revision, LogPath: logPath}); err != nil { t.Fatalf("create task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(&agentReportServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}}, availableNodeIDs: map[string]struct{}{"main": {}}, logState: &taskLogAckState{confirmedBy: make(map[string]uint64)}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor)) mux.Handle(reportPath, reportHandler) httpServer := httptest.NewUnstartedServer(mux) httpServer.EnableHTTP2 = true httpServer.StartTLS() defer httpServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) finishedAt := timestamppb.New(time.Date(2026, 4, 4, 18, 2, 0, 0, time.UTC)) if _, err := reportClient.ReportTaskState(ctx, connect.NewRequest(&agentv1.ReportTaskStateRequest{TaskId: "task-no-caddy", Status: agentv1.AgentTaskStatus_AGENT_TASK_STATUS_SUCCEEDED, FinishedAt: finishedAt})); err != nil { t.Fatalf("report task state: %v", err) }
reloadTasks, totalCount, err := db.ListTasks(ctx, []string{string(task.StatusPending)}, []string{"edge"}, []string{"main"}, []string{string(task.TypeCaddyReload)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list caddy reload tasks: %v", err) } if totalCount != 0 || len(reloadTasks) != 0 { t.Fatalf("expected no queued caddy reload task, got total=%d tasks=%+v", totalCount, reloadTasks) }}
func TestAgentReportTaskStateQueuesCaddyReloadAfterStop(t *testing.T) { t.Parallel()
db := openControllerTestDB(t) defer func() { _ = db.Close() }()
ctx := context.Background() repoDir := t.TempDir() createGitRepoWithContent(t, repoDir, map[string]string{ "demo/composia-meta.yaml": "name: demo\nnodes:\n - main\nnetwork:\n caddy:\n enabled: true\n source: ./demo.caddy\n", "edge/composia-meta.yaml": "name: edge\nnodes:\n - main\ninfra:\n caddy:\n compose_service: caddy\n config_dir: /etc/caddy\n", }) logDir := filepath.Join(t.TempDir(), "logs") if err := os.MkdirAll(filepath.Join(logDir, "tasks"), 0o750); err != nil { t.Fatalf("create task log dir: %v", err) } if err := db.SyncConfiguredNodes(ctx, []string{"main"}); err != nil { t.Fatalf("sync configured nodes: %v", err) } if err := db.RecordHeartbeat(ctx, store.NodeHeartbeat{NodeID: "main", HeartbeatAt: time.Date(2026, 4, 4, 16, 59, 0, 0, time.UTC)}); err != nil { t.Fatalf("record heartbeat: %v", err) } if err := syncDeclaredServicesForTests(ctx, db, "demo", "edge"); err != nil { t.Fatalf("sync declared services: %v", err) } revision := currentRevision(t, repoDir) paramsJSON, err := json.Marshal(serviceTaskParams{ServiceDir: "demo"}) if err != nil { t.Fatalf("marshal params: %v", err) } logPath := filepath.Join(t.TempDir(), "logs", "task.log") if _, err := db.CreateTask(ctx, task.Record{TaskID: "task-stop-caddy", Type: task.TypeStop, Source: task.SourceCLI, ServiceName: "demo", NodeID: "main", ParamsJSON: string(paramsJSON), RepoRevision: revision, LogPath: logPath}); err != nil { t.Fatalf("create task: %v", err) }
interceptor := rpcutil.NewServerBearerAuthInterceptor(func(token string) (string, error) { if token != "main-token" { return "", assertError("unexpected token") } return "main", nil }) mux := http.NewServeMux() reportPath, reportHandler := agentv1connect.NewAgentReportServiceHandler(&agentReportServer{db: db, cfg: &config.ControllerConfig{RepoDir: repoDir, LogDir: logDir, Nodes: []config.NodeConfig{{ID: "main"}}}, availableNodeIDs: map[string]struct{}{"main": {}}, logState: &taskLogAckState{confirmedBy: make(map[string]uint64)}, taskQueue: newTaskQueueNotifier(), taskResults: newTaskResultNotifier()}, connect.WithInterceptors(interceptor)) mux.Handle(reportPath, reportHandler) httpServer := httptest.NewUnstartedServer(mux) httpServer.EnableHTTP2 = true httpServer.StartTLS() defer httpServer.Close()
reportClient := agentv1connect.NewAgentReportServiceClient(httpServer.Client(), httpServer.URL, connect.WithInterceptors(rpcutil.NewStaticBearerAuthInterceptor("main-token"))) finishedAt := timestamppb.New(time.Date(2026, 4, 4, 18, 2, 0, 0, time.UTC)) if _, err := reportClient.ReportTaskState(ctx, connect.NewRequest(&agentv1.ReportTaskStateRequest{TaskId: "task-stop-caddy", Status: agentv1.AgentTaskStatus_AGENT_TASK_STATUS_SUCCEEDED, FinishedAt: finishedAt})); err != nil { t.Fatalf("report task state: %v", err) }
reloadTasks, totalCount, err := db.ListTasks(ctx, []string{string(task.StatusPending)}, []string{"edge"}, []string{"main"}, []string{string(task.TypeCaddyReload)}, nil, nil, nil, nil, 1, 10) if err != nil { t.Fatalf("list caddy reload tasks: %v", err) } if totalCount != 1 || len(reloadTasks) != 1 { t.Fatalf("expected one queued caddy reload task after stop, got total=%d tasks=%+v", totalCount, reloadTasks) } reloadDetail, err := db.GetTask(ctx, reloadTasks[0].TaskID) if err != nil { t.Fatalf("get caddy reload task: %v", err) } if reloadDetail.Record.Source != task.SourceCLI { t.Fatalf("expected caddy reload source cli, got %q", reloadDetail.Record.Source) }}