From 8a0e84b2146a4b97b4e42921f158dd6f455a6888 Mon Sep 17 00:00:00 2001 From: naiba Date: Mon, 20 Jul 2026 04:46:55 +0000 Subject: [PATCH] test(agentcompat): add dashboard runtime harness Co-authored-by: naiba/CloudCode --- .../internal/dashboard/accessors.go | 102 ++++++++ .../internal/dashboard/adversarial_test.go | 215 ++++++++++++++++ .../internal/dashboard/bootstrap.go | 127 ++++++++++ .../agentcompat/internal/dashboard/config.go | 62 +++++ .../internal/dashboard/config_test.go | 66 +++++ .../internal/dashboard/dashboard.go | 233 ++++++++++++++++++ .../internal/dashboard/dashboard_test.go | 216 ++++++++++++++++ .../agentcompat/internal/dashboard/fixture.go | 82 ++++++ .../dashboard/io_stream_state_test.go | 60 +++++ .../internal/dashboard/lifecycle.go | 156 ++++++++++++ .../internal/dashboard/receipt_lifecycle.go | 94 +++++++ .../dashboard/receipt_lifecycle_test.go | 63 +++++ .../internal/dashboard/receipt_runtime.go | 100 ++++++++ .../internal/dashboard/receipt_set.go | 114 +++++++++ .../internal/dashboard/receipt_set_test.go | 177 +++++++++++++ .../dashboard/restart_runtime_test.go | 55 +++++ .../agentcompat/internal/dashboard/runtime.go | 186 ++++++++++++++ .../agentcompat/internal/dashboard/tls.go | 46 ++++ .../dashboard/waiter_broadcast_test.go | 119 +++++++++ .../agentcompat/internal/dashboard/waiters.go | 184 ++++++++++++++ 20 files changed, 2457 insertions(+) create mode 100644 integration/agentcompat/internal/dashboard/accessors.go create mode 100644 integration/agentcompat/internal/dashboard/adversarial_test.go create mode 100644 integration/agentcompat/internal/dashboard/bootstrap.go create mode 100644 integration/agentcompat/internal/dashboard/config.go create mode 100644 integration/agentcompat/internal/dashboard/config_test.go create mode 100644 integration/agentcompat/internal/dashboard/dashboard.go create mode 100644 integration/agentcompat/internal/dashboard/dashboard_test.go create mode 100644 integration/agentcompat/internal/dashboard/fixture.go create mode 100644 integration/agentcompat/internal/dashboard/io_stream_state_test.go create mode 100644 integration/agentcompat/internal/dashboard/lifecycle.go create mode 100644 integration/agentcompat/internal/dashboard/receipt_lifecycle.go create mode 100644 integration/agentcompat/internal/dashboard/receipt_lifecycle_test.go create mode 100644 integration/agentcompat/internal/dashboard/receipt_runtime.go create mode 100644 integration/agentcompat/internal/dashboard/receipt_set.go create mode 100644 integration/agentcompat/internal/dashboard/receipt_set_test.go create mode 100644 integration/agentcompat/internal/dashboard/restart_runtime_test.go create mode 100644 integration/agentcompat/internal/dashboard/runtime.go create mode 100644 integration/agentcompat/internal/dashboard/tls.go create mode 100644 integration/agentcompat/internal/dashboard/waiter_broadcast_test.go create mode 100644 integration/agentcompat/internal/dashboard/waiters.go diff --git a/integration/agentcompat/internal/dashboard/accessors.go b/integration/agentcompat/internal/dashboard/accessors.go new file mode 100644 index 00000000..3565659a --- /dev/null +++ b/integration/agentcompat/internal/dashboard/accessors.go @@ -0,0 +1,102 @@ +//go:build linux + +package dashboard + +import ( + "context" + "errors" + "path/filepath" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +func (dashboard *Dashboard) TLSURL() string { + if dashboard.httpsAddress == "" { + return "" + } + _, port, err := splitAddress(dashboard.httpsAddress) + if err != nil { + return "" + } + return "https://localhost:" + port +} + +func (dashboard *Dashboard) TLSCACertificatePath() string { + if dashboard.tlsFixture.CAPEM() == nil { + return "" + } + return filepath.Join(dashboard.workspace.Root(), "dashboard-ca.crt") +} + +func (dashboard *Dashboard) Clients() Clients { return dashboard.clients } + +func (dashboard *Dashboard) AuthenticatedClient(token string) (*client.Client, error) { + return client.New(client.Config{BaseURL: dashboard.URL(), HTTPClient: dashboard.restHTTPClient, BearerToken: token}) +} + +func (dashboard *Dashboard) ReleaseReceipt(ctx context.Context) error { + if dashboard.receiptConn == nil { + return errors.New("receipt gate is disabled") + } + if deadline, ok := ctx.Deadline(); ok { + if err := dashboard.receiptConn.SetWriteDeadline(deadline); err != nil { + return err + } + } + _, err := dashboard.receiptConn.Write([]byte("release\n")) + return err +} + +func (dashboard *Dashboard) Bootstrap() BootstrapResult { + result := dashboard.bootstrap + result.PATScopes = append([]string(nil), result.PATScopes...) + return result +} + +func (dashboard *Dashboard) AgentSecret() string { return agentSecret } + +func (dashboard *Dashboard) ConfigPath() string { return dashboard.configPath } +func (dashboard *Dashboard) DatabasePath() string { return dashboard.databasePath } +func (dashboard *Dashboard) LogPath() string { return dashboard.logPath } +func (dashboard *Dashboard) WorkspaceRoot() string { return dashboard.workspace.Root() } + +func (dashboard *Dashboard) PID() int { + if dashboard.supervisor == nil { + return 0 + } + return dashboard.supervisor.PID() +} + +func (dashboard *Dashboard) CleanupReceipt() processharness.CleanupReceipt { + dashboard.cleanupMu.Lock() + defer dashboard.cleanupMu.Unlock() + receipt := dashboard.cleanupReceipt + receipt.Processes = append([]processharness.CleanupRecord(nil), receipt.Processes...) + return receipt +} + +func (dashboard *Dashboard) WaitForGenerationAfter(ctx context.Context, generation uint64) error { + if dashboard.receiptEvents == nil { + return errors.New("receipt gate is disabled") + } + for { + dashboard.eventMu.RLock() + notify, closed := dashboard.eventNotify, dashboard.eventClosed + dashboard.eventMu.RUnlock() + dashboard.receiptMu.RLock() + observed := dashboard.receiptGeneration > generation + dashboard.receiptMu.RUnlock() + if observed { + return nil + } + if closed { + return ErrReceiptGateClosed + } + select { + case <-notify: + case <-ctx.Done(): + return ctx.Err() + } + } +} diff --git a/integration/agentcompat/internal/dashboard/adversarial_test.go b/integration/agentcompat/internal/dashboard/adversarial_test.go new file mode 100644 index 00000000..be97344d --- /dev/null +++ b/integration/agentcompat/internal/dashboard/adversarial_test.go @@ -0,0 +1,215 @@ +//go:build linux + +package dashboard + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "os" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" + "github.com/nezhahq/nezha/integration/agentcompat/internal/testpaths" + "github.com/nezhahq/nezha/integration/agentcompat/internal/workspace" +) + +type failingDashboardSupervisor struct{} + +func (failingDashboardSupervisor) Start() error { return nil } + +func (failingDashboardSupervisor) Stop(context.Context) error { + return errors.New("injected supervisor cleanup failure") +} + +func (failingDashboardSupervisor) Exited() <-chan struct{} { + return make(chan struct{}) +} + +func (failingDashboardSupervisor) PID() int { return 0 } + +func (failingDashboardSupervisor) ProcessGroupID() int { return 0 } + +func (failingDashboardSupervisor) CleanupRecord() processharness.CleanupRecord { + return processharness.CleanupRecord{Name: "dashboard", Error: "injected supervisor cleanup failure"} +} + +func TestDashboardAdversarial_RecreatesFreshStateAfterShutdown(t *testing.T) { + // Given + first := startDashboardWithoutCleanup(t, false) + firstDatabasePath := first.DatabasePath() + firstWorkspaceRoot := first.WorkspaceRoot() + stopContext, cancel := context.WithTimeout(t.Context(), 15*time.Second) + require.NoError(t, first.Stop(stopContext)) + cancel() + + // When + second := startDashboard(t, false) + + // Then + require.NotEqual(t, firstDatabasePath, second.DatabasePath()) + require.NotEqual(t, firstWorkspaceRoot, second.WorkspaceRoot()) + require.NoDirExists(t, firstWorkspaceRoot) + require.True(t, second.Bootstrap().LoginAuthenticated) +} + +func TestDashboardAdversarial_ContextInterruptionCleansProcessAndWorkspace(t *testing.T) { + // Given + processContext, interrupt := context.WithCancel(t.Context()) + sourceDir, err := testpaths.NezhaSource(t.Name()) + require.NoError(t, err) + dashboard, err := Start(processContext, StartConfig{SourceDir: sourceDir}) + require.NoError(t, err) + root := dashboard.WorkspaceRoot() + pid := dashboard.PID() + t.Cleanup(func() { + cleanupContext, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + require.NoError(t, dashboard.Stop(cleanupContext)) + }) + + // When + interrupt() + requireDashboardCleanup(t, dashboard) + + // Then + require.NoDirExists(t, root) + require.NoFileExists(t, filepath.Join("/proc", strconv.Itoa(pid))) + require.True(t, dashboard.CleanupReceipt().Passed) + require.False(t, dashboard.CleanupReceipt().Forced) +} + +func TestDashboardAdversarial_StopDeadlineStillCleansProcessAndWorkspace(t *testing.T) { + // Given + dashboard := startDashboardWithoutCleanup(t, false) + root := dashboard.WorkspaceRoot() + pid := dashboard.PID() + stopContext, cancel := context.WithCancel(t.Context()) + cancel() + t.Cleanup(func() { + cleanupContext, cleanupCancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cleanupCancel() + require.NoError(t, dashboard.Stop(cleanupContext)) + }) + + // When + stopError := dashboard.Stop(stopContext) + + // Then + require.ErrorIs(t, stopError, context.Canceled) + requireDashboardCleanup(t, dashboard) + require.NoDirExists(t, root) + require.NoFileExists(t, filepath.Join("/proc", strconv.Itoa(pid))) + require.True(t, dashboard.CleanupReceipt().Passed) + require.False(t, dashboard.CleanupReceipt().Forced) +} + +func TestDashboardAdversarial_SupervisorErrorStillClosesWorkspace(t *testing.T) { + // Given + workspaceRoot, err := workspace.New(t.Context()) + require.NoError(t, err) + dashboard := &Dashboard{ + workspace: workspaceRoot, + supervisor: failingDashboardSupervisor{}, + cleanupDone: make(chan struct{}), + } + + // When + stopContext, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + stopError := dashboard.Stop(stopContext) + + // Then + require.ErrorContains(t, stopError, "injected supervisor cleanup failure") + require.NoDirExists(t, workspaceRoot.Root()) + require.False(t, dashboard.CleanupReceipt().Passed) +} + +func TestDashboardAdversarial_HungProcessRespectsReadinessDeadline(t *testing.T) { + // Given + workspaceParent := t.TempDir() + t.Setenv("TMPDIR", workspaceParent) + sourceDir := writeHungDashboardSource(t) + requireNoWorkspaceEntries(t, workspaceParent) + + // When + _, err := Start(t.Context(), StartConfig{ + SourceDir: sourceDir, + ReadinessTimeout: 200 * time.Millisecond, + }) + + // Then + require.ErrorContains(t, err, "dashboard login readiness") + requireNoWorkspaceEntries(t, workspaceParent) +} + +func TestDashboardAdversarial_RejectsMisleadingHTTP200Login(t *testing.T) { + // Given + dashboard := startDashboard(t, false) + requestBody := []byte(`{"username":"admin","password":"wrong-password"}`) + request, err := http.NewRequestWithContext(t.Context(), http.MethodPost, dashboard.URL()+"/api/v1/login", bytes.NewReader(requestBody)) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/json") + + // When + response, err := dashboard.restHTTPClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + responseBody, err := io.ReadAll(io.LimitReader(response.Body, 4096)) + require.NoError(t, err) + var envelope client.CommonResponse[json.RawMessage] + require.NoError(t, json.Unmarshal(responseBody, &envelope)) + + // Then + require.Equal(t, http.StatusOK, response.StatusCode) + require.False(t, envelope.Success) + require.Contains(t, envelope.Error, "Unauthorized") +} + +func writeHungDashboardSource(t *testing.T) string { + t.Helper() + sourceDir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(sourceDir, "cmd", "dashboard"), 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(sourceDir, "go.mod"), []byte("module example.com/hungdashboard\n\ngo 1.23\n"), 0o600)) + program := `package main + +import ( + "os" + "os/signal" + "syscall" +) + +func main() { + signals := make(chan os.Signal, 1) + signal.Notify(signals, syscall.SIGTERM) + <-signals +} +` + require.NoError(t, os.WriteFile(filepath.Join(sourceDir, "cmd", "dashboard", "main.go"), []byte(program), 0o600)) + return sourceDir +} + +func requireNoWorkspaceEntries(t *testing.T, workspaceParent string) { + t.Helper() + entries, err := os.ReadDir(workspaceParent) + require.NoError(t, err) + require.Empty(t, entries) +} + +func requireDashboardCleanup(t *testing.T, dashboard *Dashboard) { + t.Helper() + select { + case <-dashboard.cleanupDone: + case <-time.After(15 * time.Second): + t.Fatal("dashboard cleanup did not complete") + } +} diff --git a/integration/agentcompat/internal/dashboard/bootstrap.go b/integration/agentcompat/internal/dashboard/bootstrap.go new file mode 100644 index 00000000..43b28a90 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/bootstrap.go @@ -0,0 +1,127 @@ +//go:build linux + +package dashboard + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" + "time" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" +) + +type patRequest struct { + Name string `json:"name"` + Scopes []string `json:"scopes"` + ExpiresInDays int `json:"expires_in_days"` +} + +type patResponse struct { + ID uint64 `json:"id"` + Token string `json:"token"` + Scopes []string `json:"scopes"` +} + +func (dashboard *Dashboard) bootstrapAuthentication(ctx context.Context) (patResponse, error) { + readinessContext, cancel := context.WithTimeout(ctx, dashboard.readinessTimeout) + defer cancel() + retryTicker := time.NewTicker(dashboardRequestRetryPeriod) + defer retryTicker.Stop() + var login client.LoginResponse + var err error + for { + login, err = dashboard.clients.REST.Login(readinessContext, client.LoginRequest{Username: "admin", Password: "admin"}) + if err == nil { + break + } + select { + case <-dashboard.supervisor.Exited(): + return patResponse{}, errors.New("dashboard process exited before login readiness") + case <-readinessContext.Done(): + return patResponse{}, fmt.Errorf("dashboard login readiness: %w", errors.Join(err, readinessContext.Err())) + case <-retryTicker.C: + } + } + if login.Token == "" || login.Expire == "" { + return patResponse{}, errors.New("dashboard login response omitted JWT metadata") + } + hasJWT, hasCSRF, err := dashboard.authenticationCookies() + if err != nil { + return patResponse{}, err + } + if !hasJWT || !hasCSRF { + return patResponse{}, errors.New("dashboard login omitted authentication cookies") + } + pat, err := client.DoREST[patRequest, patResponse](readinessContext, dashboard.clients.REST, client.RESTRequest[patRequest]{ + Method: http.MethodPost, + Path: "/api/v1/api-tokens", + Body: &patRequest{ + Name: "agentcompat-admin", + Scopes: []string{"nezha:*"}, + ExpiresInDays: 0, + }, + }) + if err != nil { + return patResponse{}, fmt.Errorf("create dashboard PAT: %w", err) + } + if pat.ID == 0 || pat.Token == "" || len(pat.Scopes) != 1 || pat.Scopes[0] != "nezha:*" { + return patResponse{}, errors.New("dashboard PAT response omitted wildcard administrator access") + } + dashboard.bootstrap = BootstrapResult{ + LoginAuthenticated: true, + CSRFCookiePresent: true, + PATID: pat.ID, + PATScopes: append([]string(nil), pat.Scopes...), + } + return pat, nil +} + +func (dashboard *Dashboard) initializeAuthenticatedClients(ctx context.Context, pat patResponse) error { + config := client.Config{BaseURL: dashboard.URL(), HTTPClient: dashboard.restHTTPClient, BearerToken: pat.Token} + mcpClient, err := client.New(config) + if err != nil { + return err + } + webSocketClient, err := client.New(config) + if err != nil { + return err + } + dashboard.clients.MCP = mcpClient + dashboard.clients.WebSocket = webSocketClient + initializeResult, err := mcpClient.Initialize(ctx) + if err != nil { + return fmt.Errorf("initialize dashboard MCP: %w", err) + } + tools, err := mcpClient.ListTools(ctx) + if err != nil { + return fmt.Errorf("list dashboard MCP tools: %w", err) + } + if initializeResult.ProtocolVersion != "2024-11-05" || initializeResult.ServerInfo.Name != "nezha-mcp" || len(tools.Tools) == 0 { + return errors.New("dashboard MCP initialization returned incomplete capabilities") + } + dashboard.bootstrap.MCPProtocolVersion = initializeResult.ProtocolVersion + dashboard.bootstrap.MCPServerName = initializeResult.ServerInfo.Name + dashboard.bootstrap.MCPToolCount = len(tools.Tools) + return nil +} + +func (dashboard *Dashboard) authenticationCookies() (bool, bool, error) { + baseURL, err := url.Parse(dashboard.URL()) + if err != nil { + return false, false, fmt.Errorf("parse dashboard URL: %w", err) + } + var hasJWT bool + var hasCSRF bool + for _, cookie := range dashboard.restHTTPClient.Jar.Cookies(baseURL) { + switch cookie.Name { + case "nz-jwt": + hasJWT = cookie.Value != "" + case "nz-csrf": + hasCSRF = cookie.Value != "" + } + } + return hasJWT, hasCSRF, nil +} diff --git a/integration/agentcompat/internal/dashboard/config.go b/integration/agentcompat/internal/dashboard/config.go new file mode 100644 index 00000000..0ebd95ec --- /dev/null +++ b/integration/agentcompat/internal/dashboard/config.go @@ -0,0 +1,62 @@ +//go:build linux + +package dashboard + +import ( + "fmt" + "net" + "os" +) + +type dashboardConfig struct { + HTTPAddress string + HTTPSAddress string + ReceiptAddress string + CertificatePath string + KeyPath string +} + +func writeDashboardConfig(path string, config dashboardConfig) error { + httpHost, httpPort, err := splitAddress(config.HTTPAddress) + if err != nil { + return err + } + httpsPort := "0" + if config.HTTPSAddress != "" { + _, httpsPort, err = splitAddress(config.HTTPSAddress) + if err != nil { + return err + } + } + content := fmt.Sprintf(`listen_host: %s +listen_port: %s +location: UTC +force_auth: true +agent_secret_key: %q +jwt_timeout: 1 +enable_mcp: true +oauth2: {} +tsdb: + data_path: "" +https: + listen_port: %s + tls_cert_path: %q + tls_key_path: %q + insecure_tls: false +`, httpHost, httpPort, agentSecret, httpsPort, config.CertificatePath, config.KeyPath) + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + return fmt.Errorf("write dashboard config: %w", err) + } + return nil +} + +func splitAddress(address string) (string, string, error) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return "", "", fmt.Errorf("split loopback listener address %q: %w", address, err) + } + if host == "" || port == "" { + return "", "", fmt.Errorf("split loopback listener address %q: host and port are required", address) + } + return host, port, nil +} diff --git a/integration/agentcompat/internal/dashboard/config_test.go b/integration/agentcompat/internal/dashboard/config_test.go new file mode 100644 index 00000000..c5b4bf41 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/config_test.go @@ -0,0 +1,66 @@ +//go:build linux + +package dashboard + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestDashboardConfig_UsesDeterministicHermeticSettings(t *testing.T) { + // Given + configPath := filepath.Join(t.TempDir(), "dashboard.yaml") + + // When + err := writeDashboardConfig(configPath, dashboardConfig{ + HTTPAddress: "127.0.0.1:18008", + HTTPSAddress: "127.0.0.1:18443", + CertificatePath: "/tmp/dashboard.crt", + KeyPath: "/tmp/dashboard.key", + }) + + // Then + require.NoError(t, err) + data, err := os.ReadFile(configPath) + require.NoError(t, err) + content := string(data) + require.Contains(t, content, "listen_host: 127.0.0.1") + require.Contains(t, content, "listen_port: 18008") + require.Contains(t, content, "location: UTC") + require.Contains(t, content, "force_auth: true") + require.Contains(t, content, "enable_mcp: true") + require.Contains(t, content, "oauth2: {}") + require.Contains(t, content, "data_path: \"\"") + require.Contains(t, content, "listen_port: 18443") + require.Contains(t, content, "insecure_tls: false") + require.NotContains(t, content, jwtSecret) +} + +func TestDashboardEnvironment_RemovesAmbientNezhaOverrides(t *testing.T) { + // Given + t.Setenv("NZ_FORCEAUTH", "false") + t.Setenv("NZ_ENABLEMCP", "false") + t.Setenv("NEZHA_AGENTCOMPAT_HTTP_LISTENER_FD", "999") + + // When + environment := dashboardEnvironment(true) + + // Then + require.Contains(t, environment, "NZ_JWTSECRETKEY="+jwtSecret) + require.Contains(t, environment, "NEZHA_AGENTCOMPAT_HTTP_LISTENER_FD=3") + require.Contains(t, environment, "NEZHA_AGENTCOMPAT_HTTPS_LISTENER_FD=4") + require.NotContains(t, environment, "NZ_FORCEAUTH=false") + require.NotContains(t, environment, "NZ_ENABLEMCP=false") + require.NotContains(t, environment, "NEZHA_AGENTCOMPAT_HTTP_LISTENER_FD=999") +} + +func TestDashboardStart_RejectsRelativeSourceDirectory(t *testing.T) { + // When + _, err := Start(t.Context(), StartConfig{SourceDir: "../nezha"}) + + // Then + require.ErrorContains(t, err, "source directory must be absolute") +} diff --git a/integration/agentcompat/internal/dashboard/dashboard.go b/integration/agentcompat/internal/dashboard/dashboard.go new file mode 100644 index 00000000..e1d73e27 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/dashboard.go @@ -0,0 +1,233 @@ +//go:build linux + +package dashboard + +import ( + "bufio" + "context" + "errors" + "fmt" + "net" + "net/http" + "path/filepath" + "sync" + "time" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + "github.com/nezhahq/nezha/integration/agentcompat/internal/fixture" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" + "github.com/nezhahq/nezha/integration/agentcompat/internal/workspace" +) + +const ( + agentSecret = "0123456789abcdef0123456789abcdef" + jwtSecret = "agentcompat-dashboard-jwt-secret" + defaultReadinessTimeout = 60 * time.Second + // Active Agent streams need the Dashboard graceful shutdown window before + // the harness is allowed to escalate process-group cleanup to SIGKILL. + defaultProcessStopTimeout = 15 * time.Second + defaultProcessKillTimeout = 5 * time.Second + failedStartCleanupTimeout = 15 * time.Second + dashboardMaxLogBytes = 1 << 20 + dashboardHTTPClientTimeout = 5 * time.Second + dashboardRequestRetryPeriod = 25 * time.Millisecond +) + +type StartConfig struct { + SourceDir string + EnableTLS bool + ReceiptGate bool + ReadinessTimeout time.Duration +} + +type Clients struct { + REST *client.Client + MCP *client.Client + WebSocket *client.Client +} + +type FixtureIdentity struct { + WorkspaceRoot string + ConfigPath string + DatabasePath string + BinaryPath string + HTTP workspace.ListenerIdentity + Receipt workspace.ListenerIdentity + HTTPS workspace.ListenerIdentity +} + +type RuntimeIdentity struct { + Generation uint64 + PID int + ProcessGroupID int +} + +type dashboardGeneration struct { + supervisor dashboardSupervisor + identity RuntimeIdentity + record processharness.CleanupRecord + receiptConn net.Conn + httpTransport *http.Transport + tlsTransport *http.Transport +} + +type BootstrapResult struct { + LoginAuthenticated bool + CSRFCookiePresent bool + PATID uint64 + PATScopes []string + MCPProtocolVersion string + MCPServerName string + MCPToolCount int + TLSAuthenticated bool +} + +type dashboardSupervisor interface { + Start() error + Stop(context.Context) error + Exited() <-chan struct{} + PID() int + ProcessGroupID() int + CleanupRecord() processharness.CleanupRecord +} + +type Dashboard struct { + workspace *workspace.Workspace + supervisor dashboardSupervisor + clients Clients + restHTTPClient *http.Client + httpTransport *http.Transport + tlsTransport *http.Transport + tlsFixture fixture.LocalTLSFixture + httpAddress string + httpsAddress string + receiptAddress string + receiptConn net.Conn + receiptReader *bufio.Reader + receiptEvents chan string + eventNotify chan struct{} + eventMu sync.RWMutex + eventClosed bool + configPath string + databasePath string + logPath string + bootstrap BootstrapResult + readinessTimeout time.Duration + startConfig StartConfig + binaryPath string + generation uint64 + currentProcess *dashboardGeneration + processes []*dashboardGeneration + httpListener *workspace.OwnedListener + receiptListener *workspace.OwnedListener + httpsListener *workspace.OwnedListener + + cleanupOnce sync.Once + cleanupDone chan struct{} + cleanupMu sync.Mutex + cleanupError error + cleanupReceipt processharness.CleanupReceipt + receiptMu sync.RWMutex + receiptAccepted bool + receiptAcceptedCount uint64 + receiptGeneration uint64 + info2Mu sync.Mutex + info2Events map[string]struct{} + stateMu sync.Mutex + stateEvents map[stateEventIdentity]struct{} + mcpReceiptEvents []MCPReceiptEvent + mcpReceiptSequence uint64 + eventGeneration uint64 + lifecycleMu sync.Mutex +} + +type stateEventIdentity struct { + ServerID uint64 + UUID string + Generation uint64 + Count uint64 +} + +type MCPReceiptKind string + +const ( + MCPReceiptTask MCPReceiptKind = "task" + MCPReceiptResult MCPReceiptKind = "result" +) + +type MCPReceiptCursor struct { + Sequence uint64 +} + +type MCPReceiptEvent struct { + Sequence uint64 `json:"sequence"` + DashboardGeneration uint64 `json:"dashboard_generation"` + GateGeneration uint64 `json:"gate_generation"` + ServerID uint64 `json:"server_id"` + TaskID uint64 `json:"task_id"` + TaskType uint64 `json:"task_type"` + Kind MCPReceiptKind `json:"kind"` +} + +type MCPReceiptExpectation struct { + DashboardGeneration uint64 + GateGeneration uint64 + ServerID uint64 + TaskID uint64 + TaskType uint64 +} + +type MCPReceiptPair struct { + Task MCPReceiptEvent `json:"task"` + Result MCPReceiptEvent `json:"result"` +} + +var ErrReceiptGateClosed = errors.New("receipt gate closed") + +func Start(ctx context.Context, config StartConfig) (*Dashboard, error) { + if config.SourceDir == "" || !filepath.IsAbs(config.SourceDir) { + return nil, errors.New("dashboard source directory must be absolute") + } + // Dashboard owns cancellation order so the process group is gone before the + // workspace verifies listeners, PIDs, and temporary files are absent. + workspaceRoot, err := workspace.New(context.WithoutCancel(ctx)) + if err != nil { + return nil, fmt.Errorf("create dashboard workspace: %w", err) + } + dashboard := &Dashboard{workspace: workspaceRoot, cleanupDone: make(chan struct{}), startConfig: config} + dashboard.readinessTimeout = config.ReadinessTimeout + if dashboard.readinessTimeout <= 0 { + dashboard.readinessTimeout = defaultReadinessTimeout + } + if err := dashboard.prepare(ctx, config); err != nil { + cleanupContext, cancel := context.WithTimeout(context.WithoutCancel(ctx), failedStartCleanupTimeout) + defer cancel() + return nil, errors.Join(err, dashboard.Stop(cleanupContext)) + } + go dashboard.cleanupOnCancellation(ctx) + return dashboard, nil +} + +func (dashboard *Dashboard) Stop(ctx context.Context) error { + dashboard.cleanupOnce.Do(func() { go dashboard.cleanup(context.WithoutCancel(ctx)) }) + select { + case <-dashboard.cleanupDone: + dashboard.cleanupMu.Lock() + defer dashboard.cleanupMu.Unlock() + return dashboard.cleanupError + case <-ctx.Done(): + return ctx.Err() + } +} + +func (dashboard *Dashboard) Close(ctx context.Context) error { return dashboard.Stop(ctx) } + +func (dashboard *Dashboard) URL() string { return "http://" + dashboard.httpAddress } + +func (dashboard *Dashboard) Endpoint() string { return dashboard.httpAddress } + +func (dashboard *Dashboard) TLSEndpoint() string { return dashboard.httpsAddress } + +func (dashboard *Dashboard) ReceiptGateEnabled() bool { return dashboard.receiptAddress != "" } + +func (dashboard *Dashboard) ReceiptGateEndpoint() string { return dashboard.receiptAddress } diff --git a/integration/agentcompat/internal/dashboard/dashboard_test.go b/integration/agentcompat/internal/dashboard/dashboard_test.go new file mode 100644 index 00000000..5501a9cf --- /dev/null +++ b/integration/agentcompat/internal/dashboard/dashboard_test.go @@ -0,0 +1,216 @@ +//go:build linux + +package dashboard + +import ( + "bytes" + "context" + "crypto/x509" + "encoding/json" + "errors" + "io" + "net/http" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/golang-jwt/jwt/v4" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + "github.com/nezhahq/nezha/integration/agentcompat/internal/testpaths" + "github.com/nezhahq/nezha/model" +) + +func TestDashboard_BootstrapsSQLiteLoginPATAndMCP(t *testing.T) { + // Given + dashboard := startDashboard(t, false) + bootstrap := dashboard.Bootstrap() + + // When + database, err := gorm.Open(sqlite.Open(dashboard.DatabasePath()), &gorm.Config{}) + require.NoError(t, err) + var userCount int64 + require.NoError(t, database.Model(&model.User{}).Count(&userCount).Error) + var tokenCount int64 + require.NoError(t, database.Model(&model.APIToken{}).Count(&tokenCount).Error) + require.True(t, database.Migrator().HasTable(&model.MCPAuditLog{})) + sqlDatabase, err := database.DB() + require.NoError(t, err) + require.NoError(t, sqlDatabase.Close()) + configData, err := os.ReadFile(dashboard.ConfigPath()) + require.NoError(t, err) + logData, err := os.ReadFile(dashboard.LogPath()) + require.NoError(t, err) + jwtToken := requireJWTSignedWithDeterministicSecret(t, dashboard.Clients().REST) + unauthenticatedStatus, unauthenticatedResponse := requestUnauthenticatedInventory(t, dashboard) + + // Then + require.Equal(t, int64(1), userCount) + require.Equal(t, int64(1), tokenCount) + require.True(t, bootstrap.LoginAuthenticated) + require.True(t, bootstrap.CSRFCookiePresent) + require.NotZero(t, bootstrap.PATID) + require.Equal(t, []string{"nezha:*"}, bootstrap.PATScopes) + require.Equal(t, "2024-11-05", bootstrap.MCPProtocolVersion) + require.Equal(t, "nezha-mcp", bootstrap.MCPServerName) + require.Positive(t, bootstrap.MCPToolCount) + require.Equal(t, http.StatusOK, unauthenticatedStatus) + require.False(t, unauthenticatedResponse.Success) + require.Contains(t, unauthenticatedResponse.Error, "Unauthorized") + require.Len(t, agentSecret, 32) + require.Contains(t, string(configData), "force_auth: true") + require.Contains(t, string(configData), "enable_mcp: true") + require.Contains(t, string(configData), "agent_secret_key: \""+agentSecret+"\"") + require.NotContains(t, string(configData), jwtSecret) + require.NotContains(t, string(logData), jwtSecret) + require.NotContains(t, string(logData), agentSecret) + require.NotContains(t, string(logData), jwtToken) + require.NotContains(t, string(logData), "nzp_") + require.NotEqual(t, dashboard.ConfigPath(), dashboard.DatabasePath()) + require.FileExists(t, dashboard.DatabasePath()) + require.NotNil(t, dashboard.Clients().REST) + require.NotNil(t, dashboard.Clients().MCP) + require.NotNil(t, dashboard.Clients().WebSocket) +} + +func TestDashboard_ServesTrustedTLS(t *testing.T) { + // Given + dashboard := startDashboard(t, true) + bootstrap := dashboard.Bootstrap() + + // When + wrongHostClient, wrongHostTransport, err := dashboard.newTLSHTTPClient("wronghost.invalid") + require.NoError(t, err) + defer wrongHostTransport.CloseIdleConnections() + request, err := http.NewRequestWithContext(t.Context(), http.MethodPost, dashboard.TLSURL()+"/api/v1/login", strings.NewReader(`{"username":"admin","password":"admin"}`)) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/json") + _, err = wrongHostClient.Do(request) + + // Then + require.True(t, bootstrap.TLSAuthenticated) + require.False(t, dashboard.tlsFixture.ClientConfig("localhost").InsecureSkipVerify) + var hostnameError x509.HostnameError + require.ErrorAs(t, err, &hostnameError) +} + +func TestDashboard_RejectsWrongLogin(t *testing.T) { + // Given + dashboard := startDashboard(t, false) + + // When + _, err := dashboard.Clients().REST.Login(t.Context(), client.LoginRequest{Username: "admin", Password: "wrong-password"}) + + // Then + require.ErrorIs(t, err, client.ErrSemanticFailure) + require.ErrorContains(t, err, "Unauthorized") +} + +func TestDashboard_RejectsMalformedCSRF(t *testing.T) { + // Given + dashboard := startDashboard(t, false) + + // When + status, responseBody := postMalformedCSRF(t, dashboard) + + // Then + require.Equal(t, http.StatusForbidden, status) + var envelope client.CommonResponse[json.RawMessage] + require.NoError(t, json.Unmarshal(responseBody, &envelope)) + require.False(t, envelope.Success) + require.Contains(t, envelope.Error, "invalid CSRF token") +} + +func TestDashboard_StopsCleanly(t *testing.T) { + // Given + dashboard := startDashboardWithoutCleanup(t, false) + root := dashboard.WorkspaceRoot() + pid := dashboard.PID() + + // When + stopContext, cancel := context.WithTimeout(t.Context(), 15*time.Second) + defer cancel() + require.NoError(t, dashboard.Stop(stopContext)) + + // Then + receipt := dashboard.CleanupReceipt() + require.True(t, receipt.Passed) + require.False(t, receipt.Forced) + require.Len(t, receipt.Processes, 1) + require.NoDirExists(t, root) + require.NoFileExists(t, filepath.Join("/proc", strconv.Itoa(pid))) +} + +func startDashboard(t *testing.T, enableTLS bool) *Dashboard { + t.Helper() + dashboard := startDashboardWithoutCleanup(t, enableTLS) + t.Cleanup(func() { + stopContext, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + require.NoError(t, dashboard.Stop(stopContext)) + }) + return dashboard +} + +func startDashboardWithoutCleanup(t *testing.T, enableTLS bool) *Dashboard { + t.Helper() + sourceDir, err := testpaths.NezhaSource(t.Name()) + require.NoError(t, err) + dashboard, err := Start(t.Context(), StartConfig{SourceDir: sourceDir, EnableTLS: enableTLS}) + require.NoError(t, err) + return dashboard +} + +func postMalformedCSRF(t *testing.T, dashboard *Dashboard) (int, []byte) { + t.Helper() + requestBody, err := json.Marshal(patRequest{Name: "malformed-csrf", Scopes: []string{"nezha:*"}}) + require.NoError(t, err) + request, err := http.NewRequestWithContext(t.Context(), http.MethodPost, dashboard.URL()+"/api/v1/api-tokens", bytes.NewReader(requestBody)) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-CSRF-Token", "malformed") + response, err := dashboard.restHTTPClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + responseBody, err := io.ReadAll(io.LimitReader(response.Body, 4096)) + require.NoError(t, err) + return response.StatusCode, responseBody +} + +func requireJWTSignedWithDeterministicSecret(t *testing.T, restClient *client.Client) string { + t.Helper() + login, err := restClient.Login(t.Context(), client.LoginRequest{Username: "admin", Password: "admin"}) + require.NoError(t, err) + parsed, err := jwt.Parse(login.Token, func(token *jwt.Token) (any, error) { + if token.Method.Alg() != jwt.SigningMethodHS256.Alg() { + return nil, errors.New("unexpected JWT algorithm") + } + return []byte(jwtSecret), nil + }) + require.NoError(t, err) + require.True(t, parsed.Valid) + return login.Token +} + +func requestUnauthenticatedInventory(t *testing.T, dashboard *Dashboard) (int, client.CommonResponse[json.RawMessage]) { + t.Helper() + transport := &http.Transport{DialContext: dialAddress(dashboard.httpAddress)} + defer transport.CloseIdleConnections() + httpClient := &http.Client{Transport: transport, Timeout: dashboardHTTPClientTimeout} + request, err := http.NewRequestWithContext(t.Context(), http.MethodGet, dashboard.URL()+"/api/v1/server", nil) + require.NoError(t, err) + response, err := httpClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + responseBody, err := io.ReadAll(io.LimitReader(response.Body, 4096)) + require.NoError(t, err) + var envelope client.CommonResponse[json.RawMessage] + require.NoError(t, json.Unmarshal(responseBody, &envelope)) + return response.StatusCode, envelope +} diff --git a/integration/agentcompat/internal/dashboard/fixture.go b/integration/agentcompat/internal/dashboard/fixture.go new file mode 100644 index 00000000..4e03a265 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/fixture.go @@ -0,0 +1,82 @@ +//go:build linux + +package dashboard + +import ( + "context" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/fixture" + "github.com/nezhahq/nezha/integration/agentcompat/internal/workspace" +) + +func (dashboard *Dashboard) prepareFixture(ctx context.Context, config StartConfig) error { + fileConfig, err := dashboard.prepareListeners(config) + if err != nil { + return err + } + dashboard.configPath = filepath.Join(dashboard.workspace.Root(), "dashboard.yaml") + if err := writeDashboardConfig(dashboard.configPath, fileConfig); err != nil { + return err + } + dashboard.databasePath = filepath.Join(dashboard.workspace.Root(), "dashboard.sqlite") + dashboard.binaryPath, err = dashboard.workspace.Build(ctx, workspace.BuildSpec{Name: "dashboard", SourceDir: config.SourceDir, Package: "./cmd/dashboard", Tags: []string{"agentcompat"}}) + if err != nil { + return err + } + return nil +} + +func (dashboard *Dashboard) prepareListeners(config StartConfig) (dashboardConfig, error) { + httpListener, err := dashboard.adoptLoopbackListener() + if err != nil { + return dashboardConfig{}, err + } + dashboard.httpAddress = httpListener.Address() + dashboard.httpListener = httpListener + fileConfig := dashboardConfig{HTTPAddress: dashboard.httpAddress} + if config.ReceiptGate { + receiptListener, err := dashboard.adoptLoopbackListener() + if err != nil { + return dashboardConfig{}, err + } + dashboard.receiptAddress = receiptListener.Address() + dashboard.receiptListener = receiptListener + } + if config.EnableTLS { + if _, err := dashboard.prepareTLSListener(&fileConfig); err != nil { + return dashboardConfig{}, err + } + } + return fileConfig, nil +} + +func (dashboard *Dashboard) prepareTLSListener(config *dashboardConfig) (*os.File, error) { + tlsFixture, err := fixture.NewLocalTLSFixture(time.Now().UTC()) + if err != nil { + return nil, fmt.Errorf("generate dashboard TLS fixture: %w", err) + } + dashboard.tlsFixture = tlsFixture + config.CertificatePath = filepath.Join(dashboard.workspace.Root(), "dashboard.crt") + config.KeyPath = filepath.Join(dashboard.workspace.Root(), "dashboard.key") + if err := os.WriteFile(config.CertificatePath, tlsFixture.CertificatePEM(), 0o600); err != nil { + return nil, fmt.Errorf("write dashboard certificate: %w", err) + } + if err := os.WriteFile(filepath.Join(dashboard.workspace.Root(), "dashboard-ca.crt"), tlsFixture.CAPEM(), 0o600); err != nil { + return nil, fmt.Errorf("write dashboard CA certificate: %w", err) + } + if err := os.WriteFile(config.KeyPath, tlsFixture.PrivateKeyPEM(), 0o600); err != nil { + return nil, fmt.Errorf("write dashboard private key: %w", err) + } + listener, err := dashboard.adoptLoopbackListener() + if err != nil { + return nil, err + } + dashboard.httpsAddress = listener.Address() + dashboard.httpsListener = listener + config.HTTPSAddress = dashboard.httpsAddress + return nil, nil +} diff --git a/integration/agentcompat/internal/dashboard/io_stream_state_test.go b/integration/agentcompat/internal/dashboard/io_stream_state_test.go new file mode 100644 index 00000000..53ced697 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/io_stream_state_test.go @@ -0,0 +1,60 @@ +//go:build linux + +package dashboard + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" +) + +func TestDashboardIOStreamStateEndpointUsesPATAndRedactsStreamIdentity(t *testing.T) { + // Given + dashboard := startDashboard(t, false) + authenticated := dashboard.Clients().MCP + anonymous, err := client.New(client.Config{BaseURL: dashboard.URL()}) + require.NoError(t, err) + + // When + state, err := authenticated.IOStreamState(t.Context()) + + // Then + require.NoError(t, err) + require.Equal(t, 0, state.Count) + require.Zero(t, state.Generation) + + // When + satisfied, err := authenticated.WaitForIOStreamState(t.Context(), client.IOStreamStateExpectation{ExpectedCount: client.ExpectedIOStreamCount(0)}) + + // Then + require.NoError(t, err) + require.Equal(t, state, satisfied) + + absent, err := authenticated.WaitForIOStreamState(t.Context(), client.IOStreamStateExpectation{AbsentStreamID: "absence-only"}) + require.NoError(t, err) + require.Equal(t, state, absent) + + _, err = authenticated.WaitForIOStreamState(t.Context(), client.IOStreamStateExpectation{}) + require.ErrorIs(t, err, client.ErrSemanticFailure) + + // When + _, err = authenticated.WaitForIOStreamState(t.Context(), client.IOStreamStateExpectation{ExpectedCount: client.ExpectedIOStreamCount(-1), AbsentStreamID: "private-stream-id"}) + + // Then + require.ErrorIs(t, err, client.ErrSemanticFailure) + require.NotContains(t, err.Error(), "private-stream-id") + + // When + anonymousState, anonymousErr := anonymous.IOStreamState(t.Context()) + anonymousWait, anonymousWaitErr := anonymous.WaitForIOStreamState(t.Context(), client.IOStreamStateExpectation{ExpectedCount: client.ExpectedIOStreamCount(0)}) + + // Then + require.Error(t, anonymousErr) + require.ErrorIs(t, anonymousErr, client.ErrUnauthorized) + require.Error(t, anonymousWaitErr) + require.ErrorIs(t, anonymousWaitErr, client.ErrUnauthorized) + require.Zero(t, anonymousState) + require.Zero(t, anonymousWait) +} diff --git a/integration/agentcompat/internal/dashboard/lifecycle.go b/integration/agentcompat/internal/dashboard/lifecycle.go new file mode 100644 index 00000000..8e99ee13 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/lifecycle.go @@ -0,0 +1,156 @@ +//go:build linux + +package dashboard + +import ( + "context" + "errors" + "fmt" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +func (dashboard *Dashboard) StopProcess(ctx context.Context) (RuntimeIdentity, error) { + dashboard.lifecycleMu.Lock() + defer dashboard.lifecycleMu.Unlock() + dashboard.stateMu.Lock() + process := dashboard.currentProcess + dashboard.currentProcess = nil + dashboard.supervisor = nil + dashboard.stateMu.Unlock() + if process == nil { + return RuntimeIdentity{}, errors.New("dashboard process is not running") + } + if process.receiptConn != nil { + _ = process.receiptConn.Close() + } + if process.httpTransport != nil { + process.httpTransport.CloseIdleConnections() + } + if process.tlsTransport != nil { + process.tlsTransport.CloseIdleConnections() + } + if err := process.supervisor.Stop(ctx); err != nil { + return process.identity, fmt.Errorf("stop dashboard process: %w", err) + } + process.record = process.supervisor.CleanupRecord() + return process.identity, nil +} + +func (dashboard *Dashboard) StartProcess(ctx context.Context) (RuntimeIdentity, error) { + dashboard.lifecycleMu.Lock() + defer dashboard.lifecycleMu.Unlock() + dashboard.stateMu.Lock() + if dashboard.currentProcess != nil { + dashboard.stateMu.Unlock() + return RuntimeIdentity{}, errors.New("dashboard process is already running") + } + dashboard.generation++ + generation := dashboard.generation + dashboard.stateMu.Unlock() + dashboard.receiptMu.Lock() + dashboard.receiptAccepted = false + dashboard.receiptAcceptedCount = 0 + dashboard.receiptGeneration = 0 + dashboard.receiptMu.Unlock() + process, err := dashboard.startGeneration(ctx, generation) + if err != nil { + return RuntimeIdentity{}, err + } + dashboard.stateMu.Lock() + dashboard.currentProcess = process + dashboard.supervisor = process.supervisor + dashboard.processes = append(dashboard.processes, process) + dashboard.stateMu.Unlock() + return process.identity, nil +} + +func (dashboard *Dashboard) FixtureIdentity() FixtureIdentity { + identity := FixtureIdentity{WorkspaceRoot: dashboard.workspace.Root(), ConfigPath: dashboard.configPath, DatabasePath: dashboard.databasePath, BinaryPath: dashboard.binaryPath} + if dashboard.httpListener != nil { + identity.HTTP = dashboard.httpListener.Identity() + } + if dashboard.receiptListener != nil { + identity.Receipt = dashboard.receiptListener.Identity() + } + if dashboard.httpsListener != nil { + identity.HTTPS = dashboard.httpsListener.Identity() + } + return identity +} + +func (dashboard *Dashboard) RuntimeIdentity() RuntimeIdentity { + dashboard.stateMu.Lock() + defer dashboard.stateMu.Unlock() + if dashboard.currentProcess == nil { + return RuntimeIdentity{} + } + return dashboard.currentProcess.identity +} + +func (dashboard *Dashboard) Restart(ctx context.Context) error { + if _, err := dashboard.StopProcess(ctx); err != nil { + return err + } + _, err := dashboard.StartProcess(ctx) + return err +} + +func (dashboard *Dashboard) cleanupOnCancellation(ctx context.Context) { + select { + case <-ctx.Done(): + dashboard.cleanupOnce.Do(func() { go dashboard.cleanup(context.WithoutCancel(ctx)) }) + case <-dashboard.cleanupDone: + } +} + +func (dashboard *Dashboard) cleanup(ctx context.Context) { + defer close(dashboard.cleanupDone) + var stopError error + cleanupReceipt := dashboard.cleanupProcesses(ctx, &stopError) + if err := dashboard.workspace.Close(); err != nil { + stopError = errors.Join(stopError, fmt.Errorf("close dashboard workspace: %w", err)) + cleanupReceipt = processharness.NewCleanupReceipt(append(cleanupReceipt.Processes, processharness.CleanupRecord{Name: "dashboard-workspace", Error: client.Redact(err.Error())})) + } + dashboard.cleanupMu.Lock() + dashboard.cleanupError = stopError + dashboard.cleanupReceipt = cleanupReceipt + dashboard.cleanupMu.Unlock() +} + +func (dashboard *Dashboard) cleanupProcesses(ctx context.Context, stopError *error) processharness.CleanupReceipt { + cleanupReceipt := processharness.CleanupReceipt{} + dashboard.stateMu.Lock() + processes := append([]*dashboardGeneration(nil), dashboard.processes...) + legacySupervisor := dashboard.supervisor + dashboard.stateMu.Unlock() + if len(processes) == 0 && legacySupervisor != nil { + if err := legacySupervisor.Stop(ctx); err != nil { + *stopError = errors.Join(*stopError, fmt.Errorf("stop dashboard process: %w", err)) + } + return processharness.NewCleanupReceipt([]processharness.CleanupRecord{legacySupervisor.CleanupRecord()}) + } + for _, process := range processes { + if process.receiptConn != nil { + _ = process.receiptConn.Close() + } + if process.httpTransport != nil { + process.httpTransport.CloseIdleConnections() + } + if process.tlsTransport != nil { + process.tlsTransport.CloseIdleConnections() + } + stopContext, cancel := context.WithTimeout(ctx, failedStartCleanupTimeout) + if err := process.supervisor.Stop(stopContext); err != nil { + *stopError = errors.Join(*stopError, fmt.Errorf("stop dashboard process: %w", err)) + } + cancel() + process.record = process.supervisor.CleanupRecord() + if process.record.Forced { + *stopError = errors.Join(*stopError, errors.New("dashboard required forced SIGKILL cleanup")) + } + cleanupReceipt.Processes = append(cleanupReceipt.Processes, process.record) + } + return processharness.NewCleanupReceipt(cleanupReceipt.Processes) +} diff --git a/integration/agentcompat/internal/dashboard/receipt_lifecycle.go b/integration/agentcompat/internal/dashboard/receipt_lifecycle.go new file mode 100644 index 00000000..5cadeef4 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/receipt_lifecycle.go @@ -0,0 +1,94 @@ +//go:build linux + +package dashboard + +import ( + "context" + "errors" + "fmt" +) + +func (dashboard *Dashboard) MCPReceiptCursor() MCPReceiptCursor { + dashboard.eventMu.RLock() + defer dashboard.eventMu.RUnlock() + return MCPReceiptCursor{Sequence: dashboard.mcpReceiptSequence} +} + +func (dashboard *Dashboard) MCPReceiptEventsAfter(cursor MCPReceiptCursor) []MCPReceiptEvent { + dashboard.eventMu.RLock() + defer dashboard.eventMu.RUnlock() + events := make([]MCPReceiptEvent, 0, len(dashboard.mcpReceiptEvents)) + for _, event := range dashboard.mcpReceiptEvents { + if event.Sequence > cursor.Sequence { + events = append(events, event) + } + } + return events +} + +func (dashboard *Dashboard) WaitForMCPReceiptPairs(ctx context.Context, cursor MCPReceiptCursor, expectations []MCPReceiptExpectation) ([]MCPReceiptPair, error) { + if len(expectations) == 0 { + return nil, errors.New("MCP receipt expectations are empty") + } + for { + dashboard.eventMu.RLock() + notify, closed := dashboard.eventNotify, dashboard.eventClosed + events := append([]MCPReceiptEvent(nil), dashboard.mcpReceiptEvents...) + dashboard.eventMu.RUnlock() + pairs, complete, err := matchMCPReceiptPairs(events, cursor, expectations) + if err != nil { + return nil, err + } + if complete { + return pairs, nil + } + if closed { + return nil, ErrReceiptGateClosed + } + select { + case <-notify: + case <-ctx.Done(): + return nil, ctx.Err() + } + } +} + +func matchMCPReceiptPairs(events []MCPReceiptEvent, cursor MCPReceiptCursor, expectations []MCPReceiptExpectation) ([]MCPReceiptPair, bool, error) { + pairs := make([]MCPReceiptPair, len(expectations)) + matched := make(map[uint64]int, len(expectations)) + for _, event := range events { + if event.Sequence <= cursor.Sequence { + continue + } + index, exists := matched[event.TaskID] + if !exists { + if event.Kind != MCPReceiptTask || len(matched) >= len(expectations) { + return nil, false, fmt.Errorf("unexpected MCP receipt event after cursor: %+v", event) + } + index = len(matched) + expectation := expectations[index] + if event.ServerID != expectation.ServerID || event.TaskType != expectation.TaskType { + return nil, false, fmt.Errorf("MCP task receipt mismatch at index %d: %+v", index, event) + } + matched[event.TaskID] = index + pairs[index].Task = event + continue + } + if event.Kind != MCPReceiptResult || pairs[index].Result.TaskID != 0 { + return nil, false, fmt.Errorf("MCP task ID %d was received more than once", event.TaskID) + } + if event.ServerID != pairs[index].Task.ServerID || event.TaskType != pairs[index].Task.TaskType || event.GateGeneration != pairs[index].Task.GateGeneration || event.DashboardGeneration != pairs[index].Task.DashboardGeneration { + return nil, false, fmt.Errorf("MCP result receipt does not match task: task=%+v result=%+v", pairs[index].Task, event) + } + pairs[index].Result = event + } + if len(matched) != len(expectations) { + return nil, false, nil + } + for _, pair := range pairs { + if pair.Result.TaskID == 0 { + return nil, false, nil + } + } + return pairs, true, nil +} diff --git a/integration/agentcompat/internal/dashboard/receipt_lifecycle_test.go b/integration/agentcompat/internal/dashboard/receipt_lifecycle_test.go new file mode 100644 index 00000000..5ee93486 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/receipt_lifecycle_test.go @@ -0,0 +1,63 @@ +//go:build linux && agentcompat + +package dashboard + +import ( + "fmt" + "testing" + + "github.com/nezhahq/nezha/model" + "github.com/stretchr/testify/require" +) + +func TestMCPReceiptLifecycle_ParsesGenerationScopedTaskAndResultAfterCursor(t *testing.T) { + // Given + dashboard := &Dashboard{eventNotify: make(chan struct{}), eventGeneration: 2} + cursor := dashboard.MCPReceiptCursor() + + // When + dashboard.processReceiptLineForGeneration(2, fmt.Sprintf("task 9 7 101 %d\n", model.TaskTypeExec)) + dashboard.processReceiptLineForGeneration(2, fmt.Sprintf("result 9 7 101 %d\n", model.TaskTypeExec)) + pairs, err := dashboard.WaitForMCPReceiptPairs(t.Context(), cursor, []MCPReceiptExpectation{{ServerID: 7, TaskType: model.TaskTypeExec}}) + + // Then + require.NoError(t, err) + require.Len(t, pairs, 1) + require.Equal(t, uint64(101), pairs[0].Task.TaskID) + require.Equal(t, pairs[0].Task.TaskID, pairs[0].Result.TaskID) + require.Equal(t, uint64(2), pairs[0].Task.DashboardGeneration) + require.Equal(t, uint64(9), pairs[0].Task.GateGeneration) +} + +func TestMCPReceiptLifecycle_RejectsDuplicateTaskIDAfterCursor(t *testing.T) { + // Given + events := []MCPReceiptEvent{ + {Sequence: 1, DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec, Kind: MCPReceiptTask}, + {Sequence: 2, DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 102, TaskType: model.TaskTypeFsRead, Kind: MCPReceiptTask}, + {Sequence: 3, DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec, Kind: MCPReceiptResult}, + {Sequence: 4, DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec, Kind: MCPReceiptResult}, + } + + // When + _, _, err := matchMCPReceiptPairs(events, MCPReceiptCursor{}, []MCPReceiptExpectation{{ServerID: 7, TaskType: model.TaskTypeExec}, {ServerID: 7, TaskType: model.TaskTypeFsRead}}) + + // Then + require.Error(t, err) +} + +func TestMCPReceiptLifecycle_DiscardsStaleDashboardGeneration(t *testing.T) { + // Given + dashboard := &Dashboard{eventNotify: make(chan struct{}), eventGeneration: 2} + + // When + dashboard.processReceiptLineForGeneration(1, fmt.Sprintf("task 8 7 101 %d\n", model.TaskTypeExec)) + + // Then + require.Empty(t, dashboard.MCPReceiptEventsAfter(MCPReceiptCursor{})) +} + +func TestMCPReceiptLifecycle_DoesNotAppendOldGenerationAfterReplacement(t *testing.T) { + dashboard := &Dashboard{eventNotify: make(chan struct{}), eventGeneration: 2} + dashboard.processReceiptLineForGeneration(1, fmt.Sprintf("task 8 7 101 %d\n", model.TaskTypeExec)) + require.Empty(t, dashboard.MCPReceiptEventsAfter(MCPReceiptCursor{})) +} diff --git a/integration/agentcompat/internal/dashboard/receipt_runtime.go b/integration/agentcompat/internal/dashboard/receipt_runtime.go new file mode 100644 index 00000000..e6800653 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/receipt_runtime.go @@ -0,0 +1,100 @@ +//go:build linux + +package dashboard + +import ( + "bufio" + "fmt" + "strings" +) + +func (dashboard *Dashboard) readReceiptEvents(generation uint64, reader *bufio.Reader) { + for { + line, err := reader.ReadString('\n') + if err != nil { + dashboard.eventMu.Lock() + dashboard.stateMu.Lock() + active := dashboard.eventGeneration == generation + dashboard.stateMu.Unlock() + if !active { + dashboard.eventMu.Unlock() + return + } + dashboard.eventClosed = true + close(dashboard.eventNotify) + dashboard.eventMu.Unlock() + return + } + dashboard.processReceiptLineForGeneration(generation, line) + dashboard.eventMu.Lock() + dashboard.stateMu.Lock() + active := dashboard.eventGeneration == generation + dashboard.stateMu.Unlock() + if !active { + dashboard.eventMu.Unlock() + return + } + close(dashboard.eventNotify) + dashboard.eventNotify = make(chan struct{}) + dashboard.eventMu.Unlock() + } +} + +func (dashboard *Dashboard) processReceiptLine(line string) { + dashboard.processReceiptLineForGeneration(0, line) +} + +func (dashboard *Dashboard) processReceiptLineForGeneration(generation uint64, line string) { + if strings.HasPrefix(line, "info2 ") { + fields := strings.Fields(line) + if len(fields) == 4 { + line = fmt.Sprintf("info2 %s %s\n", fields[2], fields[3]) + } + dashboard.info2Mu.Lock() + dashboard.info2Events[line] = struct{}{} + dashboard.info2Mu.Unlock() + } + if strings.HasPrefix(line, "accepted ") { + var serverID, receiptGeneration, stateGeneration, count uint64 + var uuid string + if _, parseErr := fmt.Sscanf(line, "accepted %d %s %d %d %d", &serverID, &uuid, &receiptGeneration, &stateGeneration, &count); parseErr == nil { + dashboard.receiptMu.Lock() + dashboard.receiptAccepted = true + dashboard.receiptAcceptedCount = count + dashboard.receiptGeneration = receiptGeneration + dashboard.receiptMu.Unlock() + dashboard.stateMu.Lock() + dashboard.stateEvents[stateEventIdentity{ServerID: serverID, UUID: uuid, Generation: stateGeneration, Count: count}] = struct{}{} + dashboard.stateMu.Unlock() + } + } + if strings.HasPrefix(line, "state ") { + var serverID, generation, count uint64 + var uuid string + if _, parseErr := fmt.Sscanf(line, "state %d %s %d %d", &serverID, &uuid, &generation, &count); parseErr == nil { + dashboard.stateMu.Lock() + dashboard.stateEvents[stateEventIdentity{ServerID: serverID, UUID: uuid, Generation: generation, Count: count}] = struct{}{} + dashboard.stateMu.Unlock() + } + } + if strings.HasPrefix(line, "task ") || strings.HasPrefix(line, "result ") { + var kind string + var gateGeneration, serverID, taskID, taskType uint64 + if _, parseErr := fmt.Sscanf(line, "%s %d %d %d %d", &kind, &gateGeneration, &serverID, &taskID, &taskType); parseErr == nil { + dashboard.eventMu.Lock() + dashboard.stateMu.Lock() + if generation != 0 && dashboard.eventGeneration != generation { + dashboard.stateMu.Unlock() + dashboard.eventMu.Unlock() + return + } + dashboard.mcpReceiptSequence++ + dashboard.mcpReceiptEvents = append(dashboard.mcpReceiptEvents, MCPReceiptEvent{ + Sequence: dashboard.mcpReceiptSequence, DashboardGeneration: generation, GateGeneration: gateGeneration, + ServerID: serverID, TaskID: taskID, TaskType: taskType, Kind: MCPReceiptKind(kind), + }) + dashboard.stateMu.Unlock() + dashboard.eventMu.Unlock() + } + } +} diff --git a/integration/agentcompat/internal/dashboard/receipt_set.go b/integration/agentcompat/internal/dashboard/receipt_set.go new file mode 100644 index 00000000..f3a8f1b0 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/receipt_set.go @@ -0,0 +1,114 @@ +//go:build linux + +package dashboard + +import ( + "context" + "errors" + "fmt" +) + +func (dashboard *Dashboard) WaitForMCPReceiptSet(ctx context.Context, cursor MCPReceiptCursor, expectations []MCPReceiptExpectation) ([]MCPReceiptPair, error) { + if err := validateMCPReceiptExpectations(expectations); err != nil { + return nil, err + } + for { + dashboard.eventMu.RLock() + notify, closed := dashboard.eventNotify, dashboard.eventClosed + events := append([]MCPReceiptEvent(nil), dashboard.mcpReceiptEvents...) + dashboard.eventMu.RUnlock() + pairs, complete, err := matchMCPReceiptSet(events, cursor, expectations) + if err != nil { + return nil, err + } + if complete { + return pairs, nil + } + if closed { + return nil, ErrReceiptGateClosed + } + select { + case <-notify: + case <-ctx.Done(): + return nil, ctx.Err() + } + } +} + +type mcpReceiptIdentity struct { + dashboardGeneration uint64 + gateGeneration uint64 + serverID uint64 + taskID uint64 + taskType uint64 +} + +func validateMCPReceiptExpectations(expectations []MCPReceiptExpectation) error { + if len(expectations) == 0 { + return errors.New("MCP receipt expectations are empty") + } + seen := make(map[mcpReceiptIdentity]struct{}, len(expectations)) + for _, expectation := range expectations { + identity := mcpReceiptIdentity{dashboardGeneration: expectation.DashboardGeneration, gateGeneration: expectation.GateGeneration, serverID: expectation.ServerID, taskID: expectation.TaskID, taskType: expectation.TaskType} + // Stress evidence requires exact generation-aware identity; zero is never a wildcard. + if expectation.DashboardGeneration == 0 || expectation.GateGeneration == 0 || expectation.ServerID == 0 || expectation.TaskID == 0 || expectation.TaskType == 0 { + return fmt.Errorf("invalid MCP receipt expectation: %+v", expectation) + } + if _, duplicate := seen[identity]; duplicate { + return fmt.Errorf("duplicate MCP receipt expectation for server %d task type %d", expectation.ServerID, expectation.TaskType) + } + seen[identity] = struct{}{} + } + return nil +} + +func matchMCPReceiptSet(events []MCPReceiptEvent, cursor MCPReceiptCursor, expectations []MCPReceiptExpectation) ([]MCPReceiptPair, bool, error) { + if err := validateMCPReceiptExpectations(expectations); err != nil { + return nil, false, err + } + indices := make(map[mcpReceiptIdentity]int, len(expectations)) + for index, expectation := range expectations { + indices[mcpReceiptIdentity{dashboardGeneration: expectation.DashboardGeneration, gateGeneration: expectation.GateGeneration, serverID: expectation.ServerID, taskID: expectation.TaskID, taskType: expectation.TaskType}] = index + } + pairs := make([]MCPReceiptPair, len(expectations)) + taskIndices := make(map[uint64]int, len(expectations)) + for _, event := range events { + if event.Sequence <= cursor.Sequence { + continue + } + identity := mcpReceiptIdentity{dashboardGeneration: event.DashboardGeneration, gateGeneration: event.GateGeneration, serverID: event.ServerID, taskID: event.TaskID, taskType: event.TaskType} + index, expected := indices[identity] + if !expected { + return nil, false, fmt.Errorf("unexpected MCP receipt event after cursor: %+v", event) + } + switch event.Kind { + case MCPReceiptTask: + if _, duplicate := taskIndices[event.TaskID]; duplicate || pairs[index].Task.TaskID != 0 { + return nil, false, fmt.Errorf("duplicate MCP task receipt for server %d task type %d: %+v", event.ServerID, event.TaskType, event) + } + taskIndices[event.TaskID] = index + pairs[index].Task = event + case MCPReceiptResult: + taskIndex, exists := taskIndices[event.TaskID] + if !exists { + return nil, false, fmt.Errorf("MCP result receipt has no matching task: %+v", event) + } + if taskIndex != index || pairs[index].Result.TaskID != 0 { + return nil, false, fmt.Errorf("duplicate or mismatched MCP result receipt: %+v", event) + } + task := pairs[index].Task + if event.ServerID != task.ServerID || event.TaskType != task.TaskType || event.GateGeneration != task.GateGeneration || event.DashboardGeneration != task.DashboardGeneration || event.TaskID != task.TaskID { + return nil, false, fmt.Errorf("MCP result receipt does not match task: task=%+v result=%+v", task, event) + } + pairs[index].Result = event + default: + return nil, false, fmt.Errorf("unexpected MCP receipt kind %q: %+v", event.Kind, event) + } + } + for _, pair := range pairs { + if pair.Task.TaskID == 0 || pair.Result.TaskID == 0 { + return nil, false, nil + } + } + return pairs, true, nil +} diff --git a/integration/agentcompat/internal/dashboard/receipt_set_test.go b/integration/agentcompat/internal/dashboard/receipt_set_test.go new file mode 100644 index 00000000..c7ba94f9 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/receipt_set_test.go @@ -0,0 +1,177 @@ +//go:build linux && agentcompat + +package dashboard + +import ( + "context" + "testing" + "time" + + "github.com/nezhahq/nezha/model" + "github.com/stretchr/testify/require" +) + +func TestMCPReceiptSet_MatchesUnorderedExactServerTaskIdentities(t *testing.T) { + expectations := []MCPReceiptExpectation{ + {DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec}, + {DashboardGeneration: 2, GateGeneration: 9, ServerID: 8, TaskID: 102, TaskType: model.TaskTypeFsRead}, + } + events := []MCPReceiptEvent{ + mcpReceiptEvent(1, 2, 9, 8, 102, model.TaskTypeFsRead, MCPReceiptTask), + mcpReceiptEvent(2, 2, 9, 8, 102, model.TaskTypeFsRead, MCPReceiptResult), + mcpReceiptEvent(3, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptTask), + mcpReceiptEvent(4, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptResult), + } + + pairs, complete, err := matchMCPReceiptSet(events, MCPReceiptCursor{}, expectations) + + require.NoError(t, err) + require.True(t, complete) + require.Equal(t, uint64(7), pairs[0].Task.ServerID) + require.Equal(t, uint64(101), pairs[0].Result.TaskID) + require.Equal(t, uint64(8), pairs[1].Task.ServerID) + require.Equal(t, uint64(102), pairs[1].Result.TaskID) +} + +func TestMCPReceiptSet_RejectsMissingDuplicateAndMismatchedReceipts(t *testing.T) { + expectation := []MCPReceiptExpectation{{DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec}} + task := mcpReceiptEvent(1, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptTask) + + tests := map[string][]MCPReceiptEvent{ + "missing result": {task}, + "duplicate task identity": {task, mcpReceiptEvent(2, 2, 9, 7, 102, model.TaskTypeExec, MCPReceiptTask)}, + "duplicate task ID": {task, mcpReceiptEvent(2, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptTask)}, + "mismatched server": {mcpReceiptEvent(1, 2, 9, 8, 101, model.TaskTypeExec, MCPReceiptTask)}, + "result gate generation mismatch": {task, mcpReceiptEvent(2, 2, 10, 7, 101, model.TaskTypeExec, MCPReceiptResult)}, + "result dashboard generation mismatch": {task, mcpReceiptEvent(2, 3, 9, 7, 101, model.TaskTypeExec, MCPReceiptResult)}, + } + for name, events := range tests { + t.Run(name, func(t *testing.T) { + _, complete, err := matchMCPReceiptSet(events, MCPReceiptCursor{}, expectation) + if name == "missing result" { + require.NoError(t, err) + require.False(t, complete) + return + } + require.Error(t, err) + require.False(t, complete) + }) + } +} + +func TestMCPReceiptSet_WaitsForEventAndPreservesCursorGeneration(t *testing.T) { + dashboard := newWaiterDashboard() + dashboard.eventGeneration = 2 + dashboard.mcpReceiptEvents = []MCPReceiptEvent{ + mcpReceiptEvent(1, 1, 8, 7, 100, model.TaskTypeExec, MCPReceiptTask), + mcpReceiptEvent(2, 1, 8, 7, 100, model.TaskTypeExec, MCPReceiptResult), + } + dashboard.mcpReceiptSequence = 2 + cursor := dashboard.MCPReceiptCursor() + result := make(chan struct { + pairs []MCPReceiptPair + err error + }, 1) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + go func() { + pairs, err := dashboard.WaitForMCPReceiptSet(ctx, cursor, []MCPReceiptExpectation{{DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec}}) + result <- struct { + pairs []MCPReceiptPair + err error + }{pairs: pairs, err: err} + }() + + dashboard.eventMu.Lock() + dashboard.mcpReceiptEvents = append(dashboard.mcpReceiptEvents, + mcpReceiptEvent(3, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptTask), + mcpReceiptEvent(4, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptResult), + ) + dashboard.mcpReceiptSequence = 4 + close(dashboard.eventNotify) + dashboard.eventNotify = make(chan struct{}) + dashboard.eventMu.Unlock() + + received := <-result + require.NoError(t, received.err) + require.Len(t, received.pairs, 1) + require.Equal(t, uint64(101), received.pairs[0].Task.TaskID) + require.Equal(t, uint64(2), received.pairs[0].Task.DashboardGeneration) + require.Equal(t, uint64(9), received.pairs[0].Task.GateGeneration) +} + +func TestMCPReceiptSet_RespectsCancellationAndDeadline(t *testing.T) { + dashboard := newWaiterDashboard() + expectations := []MCPReceiptExpectation{{DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec}} + + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + _, err := dashboard.WaitForMCPReceiptSet(cancelled, MCPReceiptCursor{}, expectations) + require.ErrorIs(t, err, context.Canceled) + + expired, expire := context.WithDeadline(t.Context(), time.Now()) + defer expire() + _, err = dashboard.WaitForMCPReceiptSet(expired, MCPReceiptCursor{}, expectations) + require.ErrorIs(t, err, context.DeadlineExceeded) +} + +func TestMCPReceiptSet_RejectsDuplicateExpectations(t *testing.T) { + _, err := (&Dashboard{}).WaitForMCPReceiptSet(t.Context(), MCPReceiptCursor{}, []MCPReceiptExpectation{ + {DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec}, + {DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec}, + }) + require.Error(t, err) +} + +func TestMCPReceiptSet_RejectsWrongExpectedGenerationAndTaskID(t *testing.T) { + // Given + events := []MCPReceiptEvent{ + mcpReceiptEvent(1, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptTask), + mcpReceiptEvent(2, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptResult), + } + expectation := MCPReceiptExpectation{DashboardGeneration: 3, GateGeneration: 9, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec} + + // When + _, _, err := matchMCPReceiptSet(events, MCPReceiptCursor{}, []MCPReceiptExpectation{expectation}) + + // Then + require.Error(t, err) +} + +func TestMCPReceiptSet_RejectsZeroGateGenerationExpectation(t *testing.T) { + // Given + dashboard := newWaiterDashboard() + dashboard.eventGeneration = 2 + dashboard.mcpReceiptEvents = []MCPReceiptEvent{ + mcpReceiptEvent(1, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptTask), + mcpReceiptEvent(2, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptResult), + } + expectation := []MCPReceiptExpectation{{DashboardGeneration: 2, GateGeneration: 0, ServerID: 7, TaskID: 101, TaskType: model.TaskTypeExec}} + + // When + _, err := dashboard.WaitForMCPReceiptSet(t.Context(), MCPReceiptCursor{}, expectation) + + // Then + require.Error(t, err) +} + +func TestMCPReceiptSet_RejectsZeroTaskIDExpectation(t *testing.T) { + // Given + dashboard := newWaiterDashboard() + dashboard.eventGeneration = 2 + dashboard.mcpReceiptEvents = []MCPReceiptEvent{ + mcpReceiptEvent(1, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptTask), + mcpReceiptEvent(2, 2, 9, 7, 101, model.TaskTypeExec, MCPReceiptResult), + } + expectation := []MCPReceiptExpectation{{DashboardGeneration: 2, GateGeneration: 9, ServerID: 7, TaskID: 0, TaskType: model.TaskTypeExec}} + + // When + _, err := dashboard.WaitForMCPReceiptSet(t.Context(), MCPReceiptCursor{}, expectation) + + // Then + require.Error(t, err) +} + +func mcpReceiptEvent(sequence, dashboardGeneration, gateGeneration, serverID, taskID, taskType uint64, kind MCPReceiptKind) MCPReceiptEvent { + return MCPReceiptEvent{Sequence: sequence, DashboardGeneration: dashboardGeneration, GateGeneration: gateGeneration, ServerID: serverID, TaskID: taskID, TaskType: taskType, Kind: kind} +} diff --git a/integration/agentcompat/internal/dashboard/restart_runtime_test.go b/integration/agentcompat/internal/dashboard/restart_runtime_test.go new file mode 100644 index 00000000..f798e7bc --- /dev/null +++ b/integration/agentcompat/internal/dashboard/restart_runtime_test.go @@ -0,0 +1,55 @@ +//go:build linux + +package dashboard + +import ( + "context" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/testpaths" + "github.com/stretchr/testify/require" +) + +func TestDashboardRestart_PreservesFixtureIdentityAndCleansGenerations(t *testing.T) { + // Given + sourceDir, err := testpaths.NezhaSource(t.Name()) + require.NoError(t, err) + dashboard, err := Start(t.Context(), StartConfig{SourceDir: sourceDir, EnableTLS: true, ReceiptGate: true}) + require.NoError(t, err) + t.Cleanup(func() { + cleanupContext, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + require.NoError(t, dashboard.Close(cleanupContext)) + }) + fixture := dashboard.FixtureIdentity() + firstRuntime := dashboard.RuntimeIdentity() + require.NotZero(t, fixture.HTTP.Inode) + require.NotZero(t, fixture.HTTPS.Inode) + + // When + stopContext, cancel := context.WithTimeout(t.Context(), 15*time.Second) + defer cancel() + _, err = dashboard.StopProcess(stopContext) + require.NoError(t, err) + firstPID := firstRuntime.PID + _, err = dashboard.StartProcess(stopContext) + require.NoError(t, err) + secondRuntime := dashboard.RuntimeIdentity() + + // Then + require.Equal(t, fixture, dashboard.FixtureIdentity()) + require.NotEqual(t, firstRuntime.PID, secondRuntime.PID) + require.Greater(t, secondRuntime.Generation, firstRuntime.Generation) + require.Equal(t, fixture.HTTP.Address, dashboard.Endpoint()) + require.Equal(t, fixture.HTTP, dashboard.FixtureIdentity().HTTP) + require.Equal(t, fixture.HTTPS, dashboard.FixtureIdentity().HTTPS) + require.FileExists(t, fixture.DatabasePath) + require.FileExists(t, fixture.ConfigPath) + require.NoError(t, dashboard.Close(stopContext)) + require.Len(t, dashboard.CleanupReceipt().Processes, 2) + require.NoFileExists(t, filepath.Join("/proc", strconv.Itoa(firstPID))) + require.NoDirExists(t, fixture.WorkspaceRoot) +} diff --git a/integration/agentcompat/internal/dashboard/runtime.go b/integration/agentcompat/internal/dashboard/runtime.go new file mode 100644 index 00000000..1edd3b3a --- /dev/null +++ b/integration/agentcompat/internal/dashboard/runtime.go @@ -0,0 +1,186 @@ +//go:build linux + +package dashboard + +import ( + "bufio" + "context" + "fmt" + "net" + "net/http" + "net/http/cookiejar" + "os" + "strings" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" + "github.com/nezhahq/nezha/integration/agentcompat/internal/workspace" +) + +func (dashboard *Dashboard) prepare(ctx context.Context, config StartConfig) error { + if err := dashboard.prepareFixture(ctx, config); err != nil { + return err + } + process, err := dashboard.startGeneration(ctx, 1) + if err != nil { + return err + } + dashboard.generation = 1 + dashboard.currentProcess, dashboard.supervisor = process, process.supervisor + dashboard.processes = append(dashboard.processes, process) + return nil +} + +func (dashboard *Dashboard) startGeneration(ctx context.Context, generation uint64) (*dashboardGeneration, error) { + files := make([]*os.File, 0, 3) + for _, listener := range []*workspace.OwnedListener{dashboard.httpListener, dashboard.receiptListener, dashboard.httpsListener} { + if listener == nil { + continue + } + file, err := listener.ExtraFile() + if err != nil { + return nil, err + } + files = append(files, file) + } + logFile, err := dashboard.workspace.Log(fmt.Sprintf("dashboard-generation-%d", generation)) + if err != nil { + return nil, err + } + dashboard.logPath = logFile.Name() + supervisor := processharness.NewSupervisor(context.WithoutCancel(ctx), processharness.Spec{ + Name: "dashboard", Path: dashboard.binaryPath, Args: []string{"-c", dashboard.configPath, "-db", dashboard.databasePath}, + Env: dashboardEnvironment(dashboard.startConfig.EnableTLS, dashboard.startConfig.ReceiptGate), ExtraFiles: files, + Stdout: logFile, Stderr: logFile, MaxLogBytes: dashboardMaxLogBytes, + TerminateTimeout: defaultProcessStopTimeout, KillTimeout: defaultProcessKillTimeout, + }) + if err := supervisor.Start(); err != nil { + return nil, err + } + identity := RuntimeIdentity{Generation: generation, PID: supervisor.PID(), ProcessGroupID: supervisor.ProcessGroupID()} + process := &dashboardGeneration{supervisor: supervisor, identity: identity} + rollback := true + defer func() { + if rollback { + _ = process.supervisor.Stop(context.WithoutCancel(ctx)) + if process.receiptConn != nil { + _ = process.receiptConn.Close() + } + } + }() + if err := dashboard.workspace.TrackPID(identity.PID); err != nil { + return nil, err + } + if err := dashboard.workspace.TrackProcessGroup(identity.ProcessGroupID); err != nil { + return nil, err + } + dashboard.supervisor = supervisor + dashboard.stateMu.Lock() + dashboard.eventGeneration = generation + dashboard.stateMu.Unlock() + if dashboard.startConfig.ReceiptGate { + connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", dashboard.receiptAddress) + if err != nil { + return nil, fmt.Errorf("connect dashboard receipt gate: %w", err) + } + process.receiptConn = connection + reader := bufio.NewReader(connection) + line, err := reader.ReadString('\n') + if err != nil { + return nil, fmt.Errorf("wait for dashboard receipt gate: %w", err) + } + if line != "ready\n" { + return nil, fmt.Errorf("unexpected dashboard receipt gate handshake %q", line) + } + dashboard.eventMu.Lock() + dashboard.receiptConn = connection + dashboard.receiptReader = reader + dashboard.receiptEvents = make(chan string, 16) + dashboard.eventNotify = make(chan struct{}) + dashboard.eventClosed = false + dashboard.eventMu.Unlock() + dashboard.info2Mu.Lock() + dashboard.info2Events = make(map[string]struct{}) + dashboard.info2Mu.Unlock() + dashboard.stateMu.Lock() + dashboard.stateEvents = make(map[stateEventIdentity]struct{}) + dashboard.stateMu.Unlock() + go dashboard.readReceiptEvents(generation, reader) + } + if err := dashboard.refreshClients(ctx); err != nil { + return nil, err + } + process.httpTransport = dashboard.httpTransport + process.tlsTransport = dashboard.tlsTransport + if dashboard.startConfig.EnableTLS { + if err := dashboard.verifyTrustedTLS(ctx); err != nil { + return nil, err + } + } + rollback = false + return process, nil +} + +func (dashboard *Dashboard) adoptLoopbackListener() (*workspace.OwnedListener, error) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + return nil, fmt.Errorf("listen for dashboard: %w", err) + } + owned, err := dashboard.workspace.AdoptListener(listener) + if err != nil { + _ = listener.Close() + return nil, fmt.Errorf("adopt dashboard listener: %w", err) + } + return owned, nil +} + +func (dashboard *Dashboard) refreshClients(ctx context.Context) error { + jar, err := cookiejar.New(nil) + if err != nil { + return err + } + dashboard.httpTransport = &http.Transport{DialContext: dialAddress(dashboard.httpAddress)} + dashboard.restHTTPClient = &http.Client{Transport: dashboard.httpTransport, Jar: jar, Timeout: dashboardHTTPClientTimeout} + dashboard.clients.REST, err = client.New(client.Config{BaseURL: dashboard.URL(), HTTPClient: dashboard.restHTTPClient}) + if err != nil { + return err + } + pat, err := dashboard.bootstrapAuthentication(ctx) + if err != nil { + return err + } + return dashboard.initializeAuthenticatedClients(ctx, pat) +} + +func dashboardEnvironment(enableTLS bool, receiptGateOption ...bool) []string { + receiptGate := len(receiptGateOption) > 0 && receiptGateOption[0] + environment := make([]string, 0, len(os.Environ())+3) + for _, variable := range os.Environ() { + if strings.HasPrefix(variable, "NZ_") || strings.HasPrefix(variable, "NEZHA_AGENTCOMPAT_") { + continue + } + environment = append(environment, variable) + } + environment = append(environment, + "NZ_JWTSECRETKEY="+jwtSecret, + "NEZHA_AGENTCOMPAT_HTTP_LISTENER_FD=3", + ) + if receiptGate { + environment = append(environment, "NEZHA_AGENTCOMPAT_RECEIPT_LISTENER_FD=4") + } + if enableTLS { + fd := 4 + if receiptGate { + fd = 5 + } + environment = append(environment, fmt.Sprintf("NEZHA_AGENTCOMPAT_HTTPS_LISTENER_FD=%d", fd)) + } + return environment +} + +func dialAddress(address string) func(context.Context, string, string) (net.Conn, error) { + return func(ctx context.Context, network, _ string) (net.Conn, error) { + var dialer net.Dialer + return dialer.DialContext(ctx, network, address) + } +} diff --git a/integration/agentcompat/internal/dashboard/tls.go b/integration/agentcompat/internal/dashboard/tls.go new file mode 100644 index 00000000..2ee7d000 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/tls.go @@ -0,0 +1,46 @@ +//go:build linux + +package dashboard + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/cookiejar" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/client" +) + +func (dashboard *Dashboard) verifyTrustedTLS(ctx context.Context) error { + httpClient, transport, err := dashboard.newTLSHTTPClient("localhost") + if err != nil { + return err + } + dashboard.tlsTransport = transport + tlsClient, err := client.New(client.Config{BaseURL: dashboard.TLSURL(), HTTPClient: httpClient}) + if err != nil { + return err + } + login, err := tlsClient.Login(ctx, client.LoginRequest{Username: "admin", Password: "admin"}) + if err != nil { + return fmt.Errorf("login through trusted dashboard TLS: %w", err) + } + if login.Token == "" { + return errors.New("trusted dashboard TLS login omitted JWT") + } + dashboard.bootstrap.TLSAuthenticated = true + return nil +} + +func (dashboard *Dashboard) newTLSHTTPClient(serverName string) (*http.Client, *http.Transport, error) { + jar, err := cookiejar.New(nil) + if err != nil { + return nil, nil, fmt.Errorf("create TLS cookie jar: %w", err) + } + transport := &http.Transport{ + TLSClientConfig: dashboard.tlsFixture.ClientConfig(serverName), + DialContext: dialAddress(dashboard.httpsAddress), + } + return &http.Client{Transport: transport, Jar: jar, Timeout: dashboardHTTPClientTimeout}, transport, nil +} diff --git a/integration/agentcompat/internal/dashboard/waiter_broadcast_test.go b/integration/agentcompat/internal/dashboard/waiter_broadcast_test.go new file mode 100644 index 00000000..a870ca76 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/waiter_broadcast_test.go @@ -0,0 +1,119 @@ +//go:build linux + +package dashboard + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func newWaiterDashboard() *Dashboard { + return &Dashboard{ + receiptEvents: make(chan string), + eventNotify: make(chan struct{}), + info2Events: make(map[string]struct{}), + stateEvents: make(map[stateEventIdentity]struct{}), + } +} + +func (dashboard *Dashboard) publishTestEvent() { + dashboard.eventMu.Lock() + close(dashboard.eventNotify) + dashboard.eventNotify = make(chan struct{}) + dashboard.eventMu.Unlock() +} + +func TestDashboardWaiters_AllObserveCachedEvents(t *testing.T) { + // Given + dashboard := newWaiterDashboard() + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + waiters := []func(context.Context) error{ + func(ctx context.Context) error { return dashboard.WaitForInfo2(ctx, 7, "uuid") }, + func(ctx context.Context) error { return dashboard.WaitForInfo2(ctx, 7, "uuid") }, + func(ctx context.Context) error { return dashboard.WaitForState(ctx, 2) }, + func(ctx context.Context) error { return dashboard.WaitForState(ctx, 2) }, + func(ctx context.Context) error { return dashboard.WaitForReceiptAccepted(ctx) }, + func(ctx context.Context) error { return dashboard.WaitForReceiptAccepted(ctx) }, + } + results := make(chan error, len(waiters)) + var group sync.WaitGroup + group.Add(len(waiters)) + for _, wait := range waiters { + go func(wait func(context.Context) error) { + defer group.Done() + results <- wait(ctx) + }(wait) + } + // When + dashboard.info2Mu.Lock() + dashboard.info2Events["info2 7 uuid\n"] = struct{}{} + dashboard.info2Mu.Unlock() + dashboard.stateMu.Lock() + dashboard.stateEvents[stateEventIdentity{ServerID: 7, UUID: "uuid", Generation: 1, Count: 2}] = struct{}{} + dashboard.stateMu.Unlock() + dashboard.receiptMu.Lock() + dashboard.receiptAcceptedCount = 1 + dashboard.receiptMu.Unlock() + dashboard.publishTestEvent() + group.Wait() + + // Then + close(results) + for err := range results { + require.NoError(t, err) + } + require.NoError(t, dashboard.WaitForInfo2(ctx, 7, "uuid")) + require.NoError(t, dashboard.WaitForState(ctx, 2)) + require.NoError(t, dashboard.WaitForReceiptAccepted(ctx)) +} + +func TestDashboardWaiters_CloseWakesAllWithTypedError(t *testing.T) { + // Given + dashboard := newWaiterDashboard() + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + results := make(chan error, 4) + go func() { results <- dashboard.WaitForInfo2(ctx, 7, "uuid") }() + go func() { results <- dashboard.WaitForState(ctx, 2) }() + go func() { results <- dashboard.WaitForReceiptAccepted(ctx) }() + go func() { results <- dashboard.WaitForSecondState(ctx) }() + + // When + dashboard.eventMu.Lock() + dashboard.eventClosed = true + close(dashboard.eventNotify) + dashboard.eventNotify = make(chan struct{}) + dashboard.eventMu.Unlock() + + // Then + for index := 0; index < 4; index++ { + require.ErrorIs(t, <-results, ErrReceiptGateClosed) + } +} + +func TestDashboardWaitForStateGenerationDoesNotCrossMatchServers(t *testing.T) { + dashboard := newWaiterDashboard() + dashboard.stateMu.Lock() + dashboard.stateEvents[stateEventIdentity{ServerID: 7, UUID: "server-seven", Generation: 1, Count: 1}] = struct{}{} + dashboard.stateMu.Unlock() + + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + err := dashboard.WaitForStateGeneration(ctx, 8, "server-eight", 1, 1) + + require.ErrorIs(t, err, context.DeadlineExceeded) +} + +func TestDashboardAcceptedEventTracksReceiptAndStateGenerationsSeparately(t *testing.T) { + dashboard := newWaiterDashboard() + dashboard.receiptEvents = make(chan string) + dashboard.processReceiptLine("accepted 7 server-seven 4 9 1\n") + + require.Equal(t, uint64(4), dashboard.ReceiptGeneration()) + require.NoError(t, dashboard.WaitForStateGeneration(t.Context(), 7, "server-seven", 9, 1)) +} diff --git a/integration/agentcompat/internal/dashboard/waiters.go b/integration/agentcompat/internal/dashboard/waiters.go new file mode 100644 index 00000000..4ec858a9 --- /dev/null +++ b/integration/agentcompat/internal/dashboard/waiters.go @@ -0,0 +1,184 @@ +//go:build linux + +package dashboard + +import ( + "context" + "errors" + "fmt" + "strings" +) + +func (dashboard *Dashboard) WaitForReceiptAccepted(ctx context.Context) error { + if dashboard.receiptEvents == nil { + return errors.New("receipt gate is disabled") + } + for { + dashboard.eventMu.RLock() + closed, notify := dashboard.eventClosed, dashboard.eventNotify + dashboard.eventMu.RUnlock() + dashboard.receiptMu.RLock() + observed := dashboard.receiptAcceptedCount > 0 + dashboard.receiptMu.RUnlock() + if closed { + return ErrReceiptGateClosed + } + if observed { + return nil + } + select { + case <-notify: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +func (dashboard *Dashboard) WaitForSecondState(ctx context.Context) error { + if dashboard.receiptEvents == nil { + return errors.New("receipt gate is disabled") + } + for { + dashboard.eventMu.RLock() + closed, notify := dashboard.eventClosed, dashboard.eventNotify + dashboard.eventMu.RUnlock() + dashboard.receiptMu.RLock() + observed := dashboard.receiptAcceptedCount >= 2 + dashboard.receiptMu.RUnlock() + if closed { + return ErrReceiptGateClosed + } + if observed { + return nil + } + select { + case <-notify: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +func (dashboard *Dashboard) WaitForState(ctx context.Context, want uint64) error { + return dashboard.waitForState(ctx, 0, "", 0, want) +} + +func (dashboard *Dashboard) WaitForStateGeneration(ctx context.Context, serverID uint64, uuid string, generation, want uint64) error { + return dashboard.waitForState(ctx, serverID, uuid, generation, want) +} + +func (dashboard *Dashboard) StateGeneration(serverID uint64, uuid string) uint64 { + dashboard.stateMu.Lock() + defer dashboard.stateMu.Unlock() + var generation uint64 + for event := range dashboard.stateEvents { + if event.ServerID == serverID && event.UUID == uuid && event.Generation > generation { + generation = event.Generation + } + } + return generation +} + +func (dashboard *Dashboard) waitForState(ctx context.Context, serverID uint64, uuid string, generation, want uint64) error { + if dashboard.receiptEvents == nil { + return errors.New("receipt gate is disabled") + } + for { + dashboard.eventMu.RLock() + closed, notify := dashboard.eventClosed, dashboard.eventNotify + dashboard.eventMu.RUnlock() + dashboard.stateMu.Lock() + observed := false + for event := range dashboard.stateEvents { + if (serverID == 0 || event.ServerID == serverID) && (uuid == "" || event.UUID == uuid) && (generation == 0 || event.Generation == generation) && event.Count == want { + observed = true + break + } + } + dashboard.stateMu.Unlock() + if closed { + return ErrReceiptGateClosed + } + if observed { + return nil + } + select { + case <-notify: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +func (dashboard *Dashboard) WaitForInfo2(ctx context.Context, serverID uint64, uuid string) error { + if dashboard.receiptEvents == nil { + return errors.New("receipt gate is disabled") + } + want := fmt.Sprintf("info2 %d %s\n", serverID, uuid) + for { + dashboard.eventMu.RLock() + closed, notify := dashboard.eventClosed, dashboard.eventNotify + dashboard.eventMu.RUnlock() + dashboard.info2Mu.Lock() + _, observed := dashboard.info2Events[want] + dashboard.info2Mu.Unlock() + if closed { + return ErrReceiptGateClosed + } + if observed { + return nil + } + select { + case <-notify: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +func (dashboard *Dashboard) WaitForInfo2UUID(ctx context.Context, uuid string) (uint64, error) { + if dashboard.receiptEvents == nil { + return 0, errors.New("receipt gate is disabled") + } + for { + dashboard.eventMu.RLock() + closed, notify := dashboard.eventClosed, dashboard.eventNotify + dashboard.eventMu.RUnlock() + dashboard.info2Mu.Lock() + for event := range dashboard.info2Events { + fields := strings.Fields(event) + if len(fields) == 3 && fields[0] == "info2" && fields[2] == uuid { + var serverID uint64 + if _, err := fmt.Sscan(fields[1], &serverID); err == nil && serverID != 0 { + dashboard.info2Mu.Unlock() + return serverID, nil + } + } + } + dashboard.info2Mu.Unlock() + if closed { + return 0, ErrReceiptGateClosed + } + select { + case <-notify: + case <-ctx.Done(): + return 0, ctx.Err() + } + } +} + +func (dashboard *Dashboard) ReceiptAccepted() bool { + dashboard.receiptMu.RLock() + defer dashboard.receiptMu.RUnlock() + return dashboard.receiptAccepted +} +func (dashboard *Dashboard) ReceiptAcceptedCount() uint64 { + dashboard.receiptMu.RLock() + defer dashboard.receiptMu.RUnlock() + return dashboard.receiptAcceptedCount +} +func (dashboard *Dashboard) ReceiptGeneration() uint64 { + dashboard.receiptMu.RLock() + defer dashboard.receiptMu.RUnlock() + return dashboard.receiptGeneration +}