From ec34ee0de09a3dc9abae6fede620c140e9c90926 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Wed, 30 Sep 2026 00:58:27 +0800 Subject: [PATCH] refactor(runtime): remove historical resource adoption paths --- CONTRIBUTING.md | 6 +- .../agentdaemon/gateway/bootstrap_url_test.go | 52 ++---- internal/agentdaemon/gateway/handler.go | 13 +- services/agents-api/HOSTED-SANDBOX-MANAGER.md | 21 +++ services/agents-api/IMPLEMENTATION.md | 4 +- .../agents-api/cmd/server/daemon_bootstrap.go | 14 -- .../cmd/server/daemon_bootstrap_test.go | 45 +----- services/agents-api/cmd/server/main.go | 2 +- .../internal/db/queries/runtime_nodes.sql | 21 --- .../internal/db/sqlc/runtime_nodes.sql.go | 101 ------------ .../internal/execution/runtime_adoption.go | 118 -------------- .../execution/runtime_adoption_test.go | 117 -------------- .../internal/execution/runtime_lifecycle.go | 1 - .../agents-api/internal/execution/worker.go | 4 +- .../agents-api/internal/runtime/gateway.go | 9 +- .../internal/store/runtime_adoption.go | 137 ---------------- .../internal/store/runtime_adoption_test.go | 151 +++--------------- .../internal/store/runtime_deployment.go | 12 +- .../internal/store/runtime_deployment_test.go | 28 ++-- .../internal/store/runtime_idle_clock_test.go | 65 ++------ .../internal/store/runtime_node_deployment.go | 31 +--- .../internal/store/runtime_node_presence.go | 4 - .../internal/store/runtime_nodes.go | 11 +- .../internal/store/runtime_nodes_test.go | 42 ++--- .../store/sandbox_deployment_mutations.go | 12 +- .../store/sandbox_deployment_setup.go | 7 +- .../store/sandbox_deployment_switch_test.go | 26 +-- .../internal/store/sandbox_specification.go | 8 - .../store/sandbox_specification_store_test.go | 2 +- 29 files changed, 166 insertions(+), 898 deletions(-) delete mode 100644 services/agents-api/internal/execution/runtime_adoption.go delete mode 100644 services/agents-api/internal/execution/runtime_adoption_test.go delete mode 100644 services/agents-api/internal/store/runtime_adoption.go diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 15b0dc872..300e827dc 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -356,7 +356,11 @@ settings remain unchanged. No old label is accepted as a fallback. Historical Runtime and project-version upgrades are not supported. Preserve older installations, Runtime files, provider resources and Session history; -install the current release separately. Use this release's template builder for +install the current release separately. Startup never verifies and rebinds historical +allocations or accepts node deployments without a valid specification. Keep the +original Core responsible for unresolved resources; see the +[operator boundary](services/agents-api/HOSTED-SANDBOX-MANAGER.md#historical-installations). +Use this release's template builder for new E2B templates. Landed migrations and historical evidence stay as repository history; ordinary current-version database initialization uses the migration runner. diff --git a/internal/agentdaemon/gateway/bootstrap_url_test.go b/internal/agentdaemon/gateway/bootstrap_url_test.go index 43a88d0b5..743c584b5 100644 --- a/internal/agentdaemon/gateway/bootstrap_url_test.go +++ b/internal/agentdaemon/gateway/bootstrap_url_test.go @@ -1,47 +1,29 @@ package gateway import ( - "context" - "errors" + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/device" "net/http" "net/http/httptest" "strings" "testing" - - "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/device" ) -func TestBootstrapResolvesPublicURLOnlyAfterAuthentication(t *testing.T) { +func TestBootstrapPublishesURLOnlyAfterAuthentication(t *testing.T) { const token = "synthetic-bootstrap-credential" - for _, tc := range []struct { - name, bearer, result string - err error - status, calls int - }{ - {"selected origin", token, "wss://core.example/api/v1/agent-daemon/ws", nil, 200, 1}, - {"unavailable", token, "", errors.New("private database detail"), 503, 1}, - {"empty origin", token, "", nil, 503, 1}, - {"unauthenticated", "wrong", "", nil, 401, 0}, - } { - t.Run(tc.name, func(t *testing.T) { - calls := 0 - h := NewHandler(HandlerConfig{Registry: NewRegistry(), PublicWSURL: "ws://private-core:8091/api/v1/agent-daemon/ws", - Authenticator: NewAuthenticator(&stubRuntimeStore{ok: true, row: device.Credential{ID: "device", WorkspaceID: "tenant", Type: RuntimeTypeAgentDaemon, CredentialHash: device.HashCredential(token)}}), - ResolveWSURL: func(context.Context, AuthenticatedRuntime) (string, error) { calls++; return tc.result, tc.err }, - }) - r := httptest.NewRequest(http.MethodPost, "/agent-daemon/bootstrap", strings.NewReader(`{"device_id":"device"}`)) - r.Header.Set("Authorization", "Bearer "+tc.bearer) - w := httptest.NewRecorder() - h.Bootstrap(w, r) - if w.Code != tc.status || calls != tc.calls { - t.Fatalf("status %d, calls %d", w.Code, calls) - } - if strings.Contains(w.Body.String(), "private-") || strings.Contains(w.Body.String(), "private database detail") { - t.Fatalf("private configuration leaked: %s", w.Body.String()) - } - if tc.status == 200 && !strings.Contains(w.Body.String(), tc.result) { - t.Fatalf("selected address missing: %s", w.Body.String()) - } - }) + const publicURL = "wss://core.example/api/v1/agent-daemon/ws" + for _, valid := range []bool{true, false} { + h := NewHandler(HandlerConfig{Registry: NewRegistry(), PublicWSURL: publicURL, + Authenticator: NewAuthenticator(&stubRuntimeStore{ok: true, row: device.Credential{ID: "device", WorkspaceID: "tenant", Type: RuntimeTypeAgentDaemon, CredentialHash: device.HashCredential(token)}})}) + r := httptest.NewRequest(http.MethodPost, "/agent-daemon/bootstrap", strings.NewReader(`{"device_id":"device"}`)) + bearer, status := "wrong", http.StatusUnauthorized + if valid { + bearer, status = token, http.StatusOK + } + r.Header.Set("Authorization", "Bearer "+bearer) + w := httptest.NewRecorder() + h.Bootstrap(w, r) + if w.Code != status || strings.Contains(w.Body.String(), publicURL) != valid { + t.Fatalf("bootstrap status=%d body=%s", w.Code, w.Body.String()) + } } } diff --git a/internal/agentdaemon/gateway/handler.go b/internal/agentdaemon/gateway/handler.go index 9783d5faf..40a47a30a 100644 --- a/internal/agentdaemon/gateway/handler.go +++ b/internal/agentdaemon/gateway/handler.go @@ -41,9 +41,6 @@ type HandlerConfig struct { // response so deployments behind a TLS terminator can advertise // the externally-reachable URL. PublicWSURL string - // ResolveWSURL chooses a configured endpoint from the authenticated device's - // persisted allocation binding. Resolution failures never use a fallback URL. - ResolveWSURL func(context.Context, AuthenticatedRuntime) (string, error) // OwnerStore enables multi-pod WebSocket ownership. When set, every // successful daemon WS dial-in claims device_id -> owner_pod_id in @@ -239,18 +236,10 @@ func (h *Handler) Bootstrap(w http.ResponseWriter, r *http.Request) { writeAuthError(w, status, code, err.Error()) return } - wsURL := h.cfg.PublicWSURL - if h.cfg.ResolveWSURL != nil { - wsURL, err = h.cfg.ResolveWSURL(r.Context(), auth) - if err != nil || wsURL == "" { - writeAuthError(w, http.StatusServiceUnavailable, "bootstrap_unavailable", "Runtime connection address is unavailable") - return - } - } resp := map[string]any{ "device_id": auth.DeviceID, "workspace_id": auth.WorkspaceID, - "ws_url": wsURL, + "ws_url": h.cfg.PublicWSURL, "heartbeat_seconds": int(h.cfg.HeartbeatInterval.Seconds()), "protocol_version": proto.Version, } diff --git a/services/agents-api/HOSTED-SANDBOX-MANAGER.md b/services/agents-api/HOSTED-SANDBOX-MANAGER.md index dc6479f22..5adc6ad2a 100644 --- a/services/agents-api/HOSTED-SANDBOX-MANAGER.md +++ b/services/agents-api/HOSTED-SANDBOX-MANAGER.md @@ -399,3 +399,24 @@ reset its idle timer. An offline node or a provider without an observation sourc returns unavailable telemetry. Provider timestamps retain their existing validation, so excessive node/Core clock skew can also make a sample unavailable; suspension eligibility continues to use database time. + +## Historical installations + +Historical upgrades and resource adoption are unsupported. Core does not verify +old allocation receipts to bind them to a new node, manufacture placements for +pending Sessions, or reconnect nodes whose deployment has no valid specification. +Startup rejects an incompatible saved selection before claiming its owner epoch. +New installations use a separate database and state directory, select a deployment +specification, and enroll nodes through the current configuration protocol. + +Keep the previous Core, its database, credentials, node identities and provider +resources together until its work and cleanup are settled. Running compute, +retained snapshots, uncertain create/restore operations and unconfirmed cleanup +remain that installation's responsibility. Do not point a new installation at +those resources, erase their receipts, or treat a refused startup as cleanup. +Operators must explicitly settle them using the original installation before +retiring it. This release does not perform that operation during startup. + +Current-version reconnect, immutable placement, maintenance fencing, idempotent +receipts, cancellation and cleanup confirmation still apply. Released allocation +history remains stored; it is never rebound to another node. diff --git a/services/agents-api/IMPLEMENTATION.md b/services/agents-api/IMPLEMENTATION.md index d64dbe105..f2e7de933 100644 --- a/services/agents-api/IMPLEMENTATION.md +++ b/services/agents-api/IMPLEMENTATION.md @@ -1185,7 +1185,9 @@ All sandbox writes require the observed generation, including initial POST at zero; reject stale state before provider preparation, reset or no-op checks, and repeat it under the committing row lock. Same-provider PUT advances the target without draining execution or retiring nodes, tokens or the owner epoch. Node -providers prepare independently and keep their old qualified serving pin. A different backend, E2B team or unspecified old selection requires reset. +providers prepare independently and keep their old qualified serving pin. A different backend or E2B team requires reset. Historical selections without a +valid specification are unsupported and rejected at startup; keep their original +Core responsible for retained resources and install the current release separately. Preserve historical allocation ownership and placement, never migrate a Session. Persist immutable allocation and placement deployment generations, distinct from diff --git a/services/agents-api/cmd/server/daemon_bootstrap.go b/services/agents-api/cmd/server/daemon_bootstrap.go index 6b88d23ac..d4c0644ea 100644 --- a/services/agents-api/cmd/server/daemon_bootstrap.go +++ b/services/agents-api/cmd/server/daemon_bootstrap.go @@ -1,24 +1,10 @@ package main import ( - "context" "errors" "net/url" - - "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/gateway" ) -// webSocketURL answers daemon bootstrap. Every daemon uses the one URL derived -// from OAC_PUBLIC_URL, except a legacy embedded node's own Runtime. -func (m *managedNodes) webSocketURL(publicURL string) func(context.Context, gateway.AuthenticatedRuntime) (string, error) { - return func(ctx context.Context, auth gateway.AuthenticatedRuntime) (string, error) { - if m != nil && m.runtime != nil && auth.RuntimeNodeID != "" && auth.RuntimeNodeID == m.runtime.LocalNodeID { - return runtimeWebSocketURL(m.runtime.CoreURL) - } - return publicURL, nil - } -} - // runtimeWebSocketURL derives the daemon WebSocket URL from a Core origin or // its /api/v1 base. func runtimeWebSocketURL(coreURL string) (string, error) { diff --git a/services/agents-api/cmd/server/daemon_bootstrap_test.go b/services/agents-api/cmd/server/daemon_bootstrap_test.go index 9ded126af..f62f6155d 100644 --- a/services/agents-api/cmd/server/daemon_bootstrap_test.go +++ b/services/agents-api/cmd/server/daemon_bootstrap_test.go @@ -10,8 +10,6 @@ import ( "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/device" "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/gateway" - "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/execution" - "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" ) type bootstrapCredentialStore struct { @@ -24,29 +22,12 @@ func (s bootstrapCredentialStore) GetDeviceCredential(context.Context, string) ( CredentialHash: device.HashCredential("synthetic-token"), RuntimeNodeID: s.nodeID, RuntimeAllocationID: s.allocationID}, true, nil } -func TestBootstrapAddressFollowsAuthenticatedAllocation(t *testing.T) { - const publicURL = "wss://private-proxy.example/api/v1/agent-daemon/ws" - local := &managedNodes{runtime: &execution.RuntimeProvider{LocalNodeID: "local-node", CoreURL: "http://host.microsandbox.internal:8091/api/v1"}} - remote := &managedNodes{setup: &managedSetup{store: &setupStore{value: store.SandboxSetup{Provider: "docker", Generation: 1}}}} - remote.setup.publish(&execution.RuntimeProvider{CoreURL: "https://selected-node-entry.example/api/v1", Generation: 1}) - zero := &managedNodes{setup: &managedSetup{}} - for _, tc := range []struct { - name, node, want string - managed *managedNodes - }{ - {"embedded managed", "local-node", "ws://host.microsandbox.internal:8091/api/v1/agent-daemon/ws", local}, - {"remote managed beside embedded node", "remote-node", publicURL, local}, - {"self-hosted beside embedded node", "", publicURL, local}, - // Every placed daemon uses the one address derived from the public URL. - {"remote managed after Web setup", "remote-node", publicURL, remote}, - {"self-hosted after Web setup", "", publicURL, remote}, - {"self-hosted before Web setup", "", publicURL, zero}, - {"standalone self-hosted", "", publicURL, nil}, - } { - t.Run(tc.name, func(t *testing.T) { +func TestBootstrapAddressUsesPublicOrigin(t *testing.T) { + const publicURL = "wss://public.example/api/v1/agent-daemon/ws" + for _, node := range []string{"local-node", "remote-node", ""} { + t.Run(node, func(t *testing.T) { h := gateway.NewHandler(gateway.HandlerConfig{Registry: gateway.NewRegistry(), PublicWSURL: publicURL, - Authenticator: gateway.NewAuthenticator(bootstrapCredentialStore{nodeID: tc.node}), - ResolveWSURL: tc.managed.webSocketURL(publicURL)}) + Authenticator: gateway.NewAuthenticator(bootstrapCredentialStore{nodeID: node})}) request := httptest.NewRequest(http.MethodPost, "https://forged.example/api/v1/agent-daemon/bootstrap", strings.NewReader(`{"device_id":"runtime","node_id":"local-node","runtime_node_id":"local-node"}`)) request.Header.Set("Authorization", "Bearer synthetic-token") @@ -54,7 +35,7 @@ func TestBootstrapAddressFollowsAuthenticatedAllocation(t *testing.T) { response := httptest.NewRecorder() h.Bootstrap(response, request) var body map[string]any - if response.Code != http.StatusOK || json.Unmarshal(response.Body.Bytes(), &body) != nil || body["ws_url"] != tc.want { + if response.Code != http.StatusOK || json.Unmarshal(response.Body.Bytes(), &body) != nil || body["ws_url"] != publicURL { t.Fatalf("bootstrap status=%d body=%s", response.Code, response.Body.String()) } if len(body) != 5 { @@ -63,17 +44,3 @@ func TestBootstrapAddressFollowsAuthenticatedAllocation(t *testing.T) { }) } } - -func TestEmbeddedBootstrapDoesNotFallbackWhenInternalAddressIsInvalid(t *testing.T) { - managed := &managedNodes{runtime: &execution.RuntimeProvider{LocalNodeID: "local-node", CoreURL: "invalid"}} - h := gateway.NewHandler(gateway.HandlerConfig{Registry: gateway.NewRegistry(), PublicWSURL: "wss://public.example/api/v1/agent-daemon/ws", - Authenticator: gateway.NewAuthenticator(bootstrapCredentialStore{nodeID: "local-node"}), - ResolveWSURL: managed.webSocketURL("wss://public.example/api/v1/agent-daemon/ws")}) - request := httptest.NewRequest(http.MethodPost, "/api/v1/agent-daemon/bootstrap", strings.NewReader(`{"device_id":"runtime"}`)) - request.Header.Set("Authorization", "Bearer synthetic-token") - response := httptest.NewRecorder() - h.Bootstrap(response, request) - if response.Code != http.StatusServiceUnavailable || strings.Contains(response.Body.String(), "public.example") { - t.Fatalf("bootstrap fell back after route failure: %d %s", response.Code, response.Body.String()) - } -} diff --git a/services/agents-api/cmd/server/main.go b/services/agents-api/cmd/server/main.go index 0f284dbdc..95bdb9092 100644 --- a/services/agents-api/cmd/server/main.go +++ b/services/agents-api/cmd/server/main.go @@ -230,7 +230,7 @@ func run() error { if err != nil { return err } - daemonHandler, registry, err = runtime.NewGatewayWithURLResolver(executionStore, wsURL, managedNodes.webSocketURL(wsURL)) + daemonHandler, registry, err = runtime.NewGateway(executionStore, wsURL) if err != nil { return err } diff --git a/services/agents-api/internal/db/queries/runtime_nodes.sql b/services/agents-api/internal/db/queries/runtime_nodes.sql index c3df72dff..f21e40ed4 100644 --- a/services/agents-api/internal/db/queries/runtime_nodes.sql +++ b/services/agents-api/internal/db/queries/runtime_nodes.sql @@ -76,22 +76,6 @@ UPDATE runtime_placements SET released_at=COALESCE(released_at,clock_timestamp() WHERE environment_id IN(SELECT id FROM environments WHERE session_id=$1) AND NOT EXISTS(SELECT 1 FROM runtime_allocations a WHERE a.environment_id=runtime_placements.environment_id); --- name: AdoptRuntimePlacements :exec -INSERT INTO runtime_placements(environment_id,node_id,released_at,deployment_generation) -SELECT a.environment_id,$1,a.released_at,a.deployment_generation FROM runtime_allocations a WHERE a.provider_key=$2 AND a.state<>'released' AND a.node_id IS NULL -ON CONFLICT(environment_id) DO NOTHING; - --- name: AdoptPendingRuntimePlacements :exec -INSERT INTO runtime_placements(environment_id,node_id,deployment_generation) -SELECT e.id,$1,(SELECT generation FROM runtime_deployment) FROM environments e JOIN sessions s ON s.id=e.session_id -WHERE s.deleted_at IS NULL AND e.status='pending' AND s.configuration->'environment'->>'type'='openai_hosted' -AND NOT EXISTS(SELECT 1 FROM runtime_allocations a WHERE a.environment_id=e.id) -ON CONFLICT(environment_id) DO NOTHING; - --- name: BindLegacyRuntimeAllocations :exec -UPDATE runtime_allocations SET node_id=$1,compute_activity_at=clock_timestamp() -WHERE provider_key=$2 AND node_id IS NULL AND state<>'released'; - -- name: ListNodeRuntimeAllocations :many SELECT a.id,a.node_id,a.deployment_generation,a.observation_error,a.state,a.compute_phase,a.compute_phase_changed_at,a.initialization,a.created_at,a.environment_id,e.session_id,s.tenant_id FROM runtime_allocations a JOIN environments e ON e.id=a.environment_id JOIN sessions s ON s.id=e.session_id @@ -103,8 +87,3 @@ SELECT id,$2,sqlc.arg(generation)::bigint FROM environments WHERE session_id=$1; -- name: SetRuntimeObservation :exec UPDATE runtime_allocations SET observation_error=$4 WHERE id=$1 AND compute_revision=$2 AND state=$3 AND state<>'released'; - --- name: ListLegacyRuntimeAllocations :many -SELECT sqlc.embed(a), e.session_id, s.tenant_id, s.deleted_at -FROM runtime_allocations a JOIN environments e ON e.id=a.environment_id JOIN sessions s ON s.id=e.session_id -WHERE a.node_id IS NULL AND a.state<>'released' AND a.id>$1 ORDER BY a.id LIMIT 32 FOR UPDATE OF a; diff --git a/services/agents-api/internal/db/sqlc/runtime_nodes.sql.go b/services/agents-api/internal/db/sqlc/runtime_nodes.sql.go index 238b38cb7..cb73f908c 100644 --- a/services/agents-api/internal/db/sqlc/runtime_nodes.sql.go +++ b/services/agents-api/internal/db/sqlc/runtime_nodes.sql.go @@ -11,50 +11,6 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) -const adoptPendingRuntimePlacements = `-- name: AdoptPendingRuntimePlacements :exec -INSERT INTO runtime_placements(environment_id,node_id,deployment_generation) -SELECT e.id,$1,(SELECT generation FROM runtime_deployment) FROM environments e JOIN sessions s ON s.id=e.session_id -WHERE s.deleted_at IS NULL AND e.status='pending' AND s.configuration->'environment'->>'type'='openai_hosted' -AND NOT EXISTS(SELECT 1 FROM runtime_allocations a WHERE a.environment_id=e.id) -ON CONFLICT(environment_id) DO NOTHING -` - -func (q *Queries) AdoptPendingRuntimePlacements(ctx context.Context, nodeID pgtype.UUID) error { - _, err := q.db.Exec(ctx, adoptPendingRuntimePlacements, nodeID) - return err -} - -const adoptRuntimePlacements = `-- name: AdoptRuntimePlacements :exec -INSERT INTO runtime_placements(environment_id,node_id,released_at,deployment_generation) -SELECT a.environment_id,$1,a.released_at,a.deployment_generation FROM runtime_allocations a WHERE a.provider_key=$2 AND a.state<>'released' AND a.node_id IS NULL -ON CONFLICT(environment_id) DO NOTHING -` - -type AdoptRuntimePlacementsParams struct { - NodeID pgtype.UUID `json:"node_id"` - ProviderKey pgtype.UUID `json:"provider_key"` -} - -func (q *Queries) AdoptRuntimePlacements(ctx context.Context, arg AdoptRuntimePlacementsParams) error { - _, err := q.db.Exec(ctx, adoptRuntimePlacements, arg.NodeID, arg.ProviderKey) - return err -} - -const bindLegacyRuntimeAllocations = `-- name: BindLegacyRuntimeAllocations :exec -UPDATE runtime_allocations SET node_id=$1,compute_activity_at=clock_timestamp() -WHERE provider_key=$2 AND node_id IS NULL AND state<>'released' -` - -type BindLegacyRuntimeAllocationsParams struct { - NodeID pgtype.UUID `json:"node_id"` - ProviderKey pgtype.UUID `json:"provider_key"` -} - -func (q *Queries) BindLegacyRuntimeAllocations(ctx context.Context, arg BindLegacyRuntimeAllocationsParams) error { - _, err := q.db.Exec(ctx, bindLegacyRuntimeAllocations, arg.NodeID, arg.ProviderKey) - return err -} - const connectRuntimeNode = `-- name: ConnectRuntimeNode :execrows UPDATE runtime_nodes SET connection_id=$2,provider_ready=false,connected_epoch=d.owner_epoch,last_seen_at=clock_timestamp() FROM runtime_deployment d WHERE runtime_nodes.id=$1 AND removed_at IS NULL AND runtime_nodes.installation_id=d.installation_id AND d.owner_epoch=$3 @@ -377,63 +333,6 @@ func (q *Queries) InsertRuntimeNode(ctx context.Context, arg InsertRuntimeNodePa return i, err } -const listLegacyRuntimeAllocations = `-- name: ListLegacyRuntimeAllocations :many -SELECT a.id, a.environment_id, a.device_id, a.provider_key, a.state, a.create_settled, a.created_at, a.kept_at, a.released_at, a.initialization, a.compute_phase, a.compute_revision, a.compute_state, a.compute_activity_at, a.compute_wake_requested, a.compute_retained_until, a.node_id, a.observation_error, a.compute_phase_changed_at, a.deployment_generation, e.session_id, s.tenant_id, s.deleted_at -FROM runtime_allocations a JOIN environments e ON e.id=a.environment_id JOIN sessions s ON s.id=e.session_id -WHERE a.node_id IS NULL AND a.state<>'released' AND a.id>$1 ORDER BY a.id LIMIT 32 FOR UPDATE OF a -` - -type ListLegacyRuntimeAllocationsRow struct { - RuntimeAllocation RuntimeAllocation `json:"runtime_allocation"` - SessionID pgtype.UUID `json:"session_id"` - TenantID pgtype.UUID `json:"tenant_id"` - DeletedAt pgtype.Timestamptz `json:"deleted_at"` -} - -func (q *Queries) ListLegacyRuntimeAllocations(ctx context.Context, id pgtype.UUID) ([]ListLegacyRuntimeAllocationsRow, error) { - rows, err := q.db.Query(ctx, listLegacyRuntimeAllocations, id) - if err != nil { - return nil, err - } - defer rows.Close() - items := []ListLegacyRuntimeAllocationsRow{} - for rows.Next() { - var i ListLegacyRuntimeAllocationsRow - if err := rows.Scan( - &i.RuntimeAllocation.ID, - &i.RuntimeAllocation.EnvironmentID, - &i.RuntimeAllocation.DeviceID, - &i.RuntimeAllocation.ProviderKey, - &i.RuntimeAllocation.State, - &i.RuntimeAllocation.CreateSettled, - &i.RuntimeAllocation.CreatedAt, - &i.RuntimeAllocation.KeptAt, - &i.RuntimeAllocation.ReleasedAt, - &i.RuntimeAllocation.Initialization, - &i.RuntimeAllocation.ComputePhase, - &i.RuntimeAllocation.ComputeRevision, - &i.RuntimeAllocation.ComputeState, - &i.RuntimeAllocation.ComputeActivityAt, - &i.RuntimeAllocation.ComputeWakeRequested, - &i.RuntimeAllocation.ComputeRetainedUntil, - &i.RuntimeAllocation.NodeID, - &i.RuntimeAllocation.ObservationError, - &i.RuntimeAllocation.ComputePhaseChangedAt, - &i.RuntimeAllocation.DeploymentGeneration, - &i.SessionID, - &i.TenantID, - &i.DeletedAt, - ); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil -} - const listNodeRuntimeAllocations = `-- name: ListNodeRuntimeAllocations :many SELECT a.id,a.node_id,a.deployment_generation,a.observation_error,a.state,a.compute_phase,a.compute_phase_changed_at,a.initialization,a.created_at,a.environment_id,e.session_id,s.tenant_id FROM runtime_allocations a JOIN environments e ON e.id=a.environment_id JOIN sessions s ON s.id=e.session_id diff --git a/services/agents-api/internal/execution/runtime_adoption.go b/services/agents-api/internal/execution/runtime_adoption.go deleted file mode 100644 index 7a425f8f1..000000000 --- a/services/agents-api/internal/execution/runtime_adoption.go +++ /dev/null @@ -1,118 +0,0 @@ -package execution - -import ( - "context" - "encoding/json" - "errors" - - "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/sandbox" - "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" -) - -// VerifyLegacyRuntimeOwnership uses only positive, read-only Provider observations. -// It runs before node placement and never becomes an execution fallback route. -func VerifyLegacyRuntimeOwnership(ctx context.Context, p sandbox.SandboxProvider, allocation store.RuntimeAllocation) (err error) { - failure := "resource_unconfirmed" - defer func() { - if err == nil { - return - } - reason := failure - switch { - case errors.Is(err, sandbox.ErrOwnership), errors.Is(err, sandbox.ErrInvalid): - reason = "ownership_mismatch" - case errors.Is(err, sandbox.ErrNotFound): - reason = "resource_missing" - case failure == "snapshot_unconfirmed": - case errors.Is(err, sandbox.ErrComputeUnconfirmed): - default: - reason = "provider_unavailable" - } - err = store.RuntimeOwnershipFailure(reason) - }() - reference := runtimeReference(allocation) - if allocation.ComputePhase == "disabled" { - if p == nil { - return sandbox.ErrComputeUnconfirmed - } - info, err := p.GetInfo(ctx, reference) - if err != nil { - return err - } - if info.Reference != reference || info.ProviderID == "" { - return sandbox.ErrOwnership - } - return nil - } - checkpoint, ok := p.(sandbox.CheckpointProvider) - if !ok { - return sandbox.ErrComputeUnconfirmed - } - var receipt runtimeCompute - if json.Unmarshal(allocation.ComputeState, &receipt) != nil || receipt.Current.ID == "" { - return sandbox.ErrOwnership - } - // A consumed restore no longer retains the source's complete compute receipt. - // Finish that transition with the previous Core instead of reconstructing it. - if snapshot := receipt.Snapshot; snapshot != nil && (snapshot.SourceID != receipt.Current.ID || snapshot.SourceName != receipt.Current.Name || snapshot.SourceGeneration != receipt.Current.Generation) { - return sandbox.ErrComputeUnconfirmed - } - found := false - inspect := func(compute sandbox.Compute) error { - observed, err := checkpoint.GetCompute(ctx, reference, compute) - if errors.Is(err, sandbox.ErrNotFound) { - return nil - } - if err != nil { - return err - } - if observed.Compute.ID == "" || !sameAdoptionCompute(compute, observed.Compute) { - return sandbox.ErrOwnership - } - found = true - return nil - } - if err := inspect(receipt.Current); err != nil { - return err - } - if receipt.Target != nil { - if err := inspect(*receipt.Target); err != nil { - return err - } - } - if receipt.Snapshot != nil || receipt.SuspendID != "" { - failure = "snapshot_unconfirmed" - // ObserveOnly verifies an already authorized operation and its full artifact. - // The helper never pauses, captures, kills or restores on this path. - var observed sandbox.ComputeState - var err error - observed, err = checkpoint.Suspend(ctx, sandbox.SuspendRequest{Reference: reference, OperationID: receipt.SuspendID, Source: receipt.Current, Snapshot: receipt.Snapshot, ObserveOnly: true}) - if err != nil { - return err - } - if !sameAdoptionCompute(receipt.Current, observed.Compute) { - return sandbox.ErrOwnership - } - if receipt.Snapshot != nil && (observed.Snapshot == nil || *receipt.Snapshot != *observed.Snapshot) { - return sandbox.ErrOwnership - } - if observed.Snapshot != nil { - snapshot := observed.Snapshot - if snapshot.OperationID != receipt.SuspendID || snapshot.SourceID != receipt.Current.ID || snapshot.SourceName != receipt.Current.Name || snapshot.SourceGeneration != receipt.Current.Generation { - return sandbox.ErrOwnership - } - found = true - } - } - if !found { - return sandbox.ErrNotFound - } - return nil -} - -func sameAdoptionCompute(want, got sandbox.Compute) bool { - if want.Name != got.Name || want.Generation != got.Generation || (want.ID != "" && want.ID != got.ID) || (want.RestoredFrom == nil) != (got.RestoredFrom == nil) { - return false - } - return want.RestoredFrom == nil || *want.RestoredFrom == *got.RestoredFrom -} diff --git a/services/agents-api/internal/execution/runtime_adoption_test.go b/services/agents-api/internal/execution/runtime_adoption_test.go deleted file mode 100644 index 68804b0d9..000000000 --- a/services/agents-api/internal/execution/runtime_adoption_test.go +++ /dev/null @@ -1,117 +0,0 @@ -package execution - -import ( - "context" - "encoding/json" - "errors" - "testing" - - "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/sandbox" - "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" -) - -// Unimplemented mutation methods panic: adoption may only make these observations. -type adoptionObserver struct { - sandbox.CheckpointProvider - info func(sandbox.Reference) (sandbox.Info, error) - compute func(sandbox.Compute) (sandbox.ComputeState, error) - snapshot func(sandbox.SuspendRequest) (sandbox.ComputeState, error) -} - -func (p adoptionObserver) GetInfo(_ context.Context, r sandbox.Reference) (sandbox.Info, error) { - return p.info(r) -} -func (p adoptionObserver) GetCompute(_ context.Context, _ sandbox.Reference, c sandbox.Compute) (sandbox.ComputeState, error) { - return p.compute(c) -} -func (p adoptionObserver) Suspend(_ context.Context, q sandbox.SuspendRequest) (sandbox.ComputeState, error) { - return p.snapshot(q) -} - -func TestLegacyOwnershipBaseRequiresPositiveIdentity(t *testing.T) { - for _, state := range []string{"running", "exited", "missing", "foreign", "unavailable"} { - t.Run(state, func(t *testing.T) { - p := adoptionObserver{info: func(r sandbox.Reference) (sandbox.Info, error) { - switch state { - case "missing": - return sandbox.Info{}, sandbox.ErrNotFound - case "foreign": - return sandbox.Info{}, sandbox.ErrOwnership - case "unavailable": - return sandbox.Info{}, errors.New("private endpoint and credentials") - } - return sandbox.Info{Reference: r, ProviderID: "exact-container", State: state}, nil - }} - err := VerifyLegacyRuntimeOwnership(t.Context(), p, store.RuntimeAllocation{ComputePhase: "disabled"}) - expected := map[string]string{"missing": "resource_missing", "foreign": "ownership_mismatch", "unavailable": "provider_unavailable"}[state] - if expected == "" { - if err != nil { - t.Fatal(err) - } - } else if err == nil || err.Error() != expected { - t.Fatal(err) - } - }) - } -} -func TestLegacyOwnershipCheckpointEvidenceMatrix(t *testing.T) { - for _, kind := range []string{"running", "stopped", "absent", "snapshot_only", "target_absent", "current_foreign", "target_foreign", "snapshot_corrupt", "snapshot_mismatch", "snapshot_missing", "consumed", "consumed_source_foreign", "consumed_later_generation"} { - t.Run(kind, func(t *testing.T) { - current := sandbox.Compute{Name: "source", ID: "source-id"} - snapshot := sandbox.SnapshotIdentity{Reference: "artifact", ID: "snapshot-id", Digest: "digest", CheckpointID: "checkpoint", CheckpointRoot: "root", OperationID: "capture", SourceName: current.Name, SourceID: current.ID} - target := sandbox.Compute{Name: "target", ID: "target-id", Generation: 1, RestoredFrom: &snapshot} - receipt := runtimeCompute{Current: current} - if kind != "running" && kind != "stopped" && kind != "absent" { - receipt.Snapshot = &snapshot - receipt.SuspendID = "capture" - } - if kind == "target_foreign" || kind == "target_absent" { - receipt.Target = &target - } - if kind == "consumed" || kind == "consumed_source_foreign" || kind == "consumed_later_generation" { - receipt.Current = target - receipt.RestoreID = "restore" - if kind == "consumed_later_generation" { - snapshot.SourceGeneration = 1 - } - } - p := adoptionObserver{ - compute: func(c sandbox.Compute) (sandbox.ComputeState, error) { - if kind == "absent" || kind == "snapshot_only" || (kind == "target_absent" && c.Name == target.Name) || (kind == "consumed" && c.Name == current.Name) { - return sandbox.ComputeState{}, sandbox.ErrNotFound - } - if (kind == "current_foreign" && c.Name == current.Name) || (kind == "target_foreign" && c.Name == target.Name) || (kind == "consumed_source_foreign" && c.Name == current.Name) { - return sandbox.ComputeState{}, sandbox.ErrOwnership - } - return sandbox.ComputeState{Compute: c, Status: kind}, nil - }, - snapshot: func(q sandbox.SuspendRequest) (sandbox.ComputeState, error) { - if !q.ObserveOnly || q.OperationID != "capture" || q.Source != current { - t.Fatal("mutating or wrong capture observation", q) - } - if kind == "snapshot_corrupt" || kind == "snapshot_missing" { - return sandbox.ComputeState{}, sandbox.ErrComputeUnconfirmed - } - got := snapshot - if kind == "snapshot_mismatch" { - got.Digest = "different" - } - return sandbox.ComputeState{Compute: current, Status: "suspended", Snapshot: &got, SourceStopped: true}, nil - }, - } - raw, err := json.Marshal(receipt) - if err != nil { - t.Fatal(err) - } - err = VerifyLegacyRuntimeOwnership(t.Context(), p, store.RuntimeAllocation{ComputePhase: "suspended", ComputeState: raw}) - expected := map[string]string{"absent": "resource_missing", "current_foreign": "ownership_mismatch", "target_foreign": "ownership_mismatch", "snapshot_corrupt": "snapshot_unconfirmed", "snapshot_missing": "snapshot_unconfirmed", "snapshot_mismatch": "ownership_mismatch", "consumed": "resource_unconfirmed", "consumed_source_foreign": "resource_unconfirmed", "consumed_later_generation": "resource_unconfirmed"}[kind] - if expected == "" { - if err != nil { - t.Fatal(err) - } - } else if err == nil || err.Error() != expected { - t.Fatal(err) - } - }) - } -} diff --git a/services/agents-api/internal/execution/runtime_lifecycle.go b/services/agents-api/internal/execution/runtime_lifecycle.go index 113857c08..3ed19c542 100644 --- a/services/agents-api/internal/execution/runtime_lifecycle.go +++ b/services/agents-api/internal/execution/runtime_lifecycle.go @@ -27,7 +27,6 @@ type RuntimeProvider struct { Mode string loadDeployment func(context.Context) (*RuntimeProvider, error) prepareDeployment RuntimeDeploymentPreparer - VerifyLegacyOwnership store.RuntimeOwnershipVerifier ProviderKind string LocalNodeID string LocalCredentialSHA256 string diff --git a/services/agents-api/internal/execution/worker.go b/services/agents-api/internal/execution/worker.go index 57d3fabfa..46278d140 100644 --- a/services/agents-api/internal/execution/worker.go +++ b/services/agents-api/internal/execution/worker.go @@ -48,10 +48,8 @@ func StartWorker(ctx context.Context, dispatcher *Dispatcher) (*Worker, error) { return nil, err } var deployment *store.RuntimeDeployment - var verify store.RuntimeOwnershipVerifier if worker.runtimes != nil && worker.runtimes.loadDeployment == nil { config := worker.runtimes.config - verify = config.VerifyLegacyOwnership deployment = &store.RuntimeDeployment{ProviderKind: config.ProviderKind, LocalNodeID: config.LocalNodeID, LocalCredentialSHA256: config.LocalCredentialSHA256, LocalMaxActive: config.LocalMaxActive, LocalMaxRetained: config.LocalMaxRetained, InstallationID: config.InstallationID, BackendFingerprint: config.BackendFingerprint, AdmissionPaused: config.AdmissionPaused} } if worker.runtimes != nil && worker.runtimes.loadDeployment != nil { @@ -60,7 +58,7 @@ func StartWorker(ctx context.Context, dispatcher *Dispatcher) (*Worker, error) { _, err = worker.runtimes.ensureDeployment(ctx) } } else { - err = owned.Store.ConfigureRuntimeDeployment(ctx, deployment, verify) + err = owned.Store.ConfigureRuntimeDeployment(ctx, deployment) } if err != nil { if worker.runtimes != nil { diff --git a/services/agents-api/internal/runtime/gateway.go b/services/agents-api/internal/runtime/gateway.go index 053935a15..e6de36d37 100644 --- a/services/agents-api/internal/runtime/gateway.go +++ b/services/agents-api/internal/runtime/gateway.go @@ -2,7 +2,6 @@ package runtime import ( - "context" "errors" "net/http" "net/url" @@ -20,12 +19,6 @@ type DeviceStore interface { // NewGateway serves the V1 daemon executor transport for both managed and // user-managed Runtime. Its credentials never grant public Session API access. func NewGateway(s DeviceStore, publicWSURL string) (http.Handler, *gateway.Registry, error) { - return NewGatewayWithURLResolver(s, publicWSURL, nil) -} - -// NewGatewayWithURLResolver chooses the bootstrap address using authenticated -// allocation bindings without replacing the gateway or its live connections. -func NewGatewayWithURLResolver(s DeviceStore, publicWSURL string, resolve func(context.Context, gateway.AuthenticatedRuntime) (string, error)) (http.Handler, *gateway.Registry, error) { u, err := url.Parse(publicWSURL) if err != nil || s == nil || (u.Scheme != "ws" && u.Scheme != "wss") || u.Hostname() == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" || u.Path != "/api/v1/agent-daemon/ws" { return nil, nil, errors.New("daemon URL must be an absolute ws(s) URL ending in /api/v1/agent-daemon/ws") @@ -33,7 +26,7 @@ func NewGatewayWithURLResolver(s DeviceStore, publicWSURL string, resolve func(c registry := gateway.NewRegistry() h := gateway.NewHandler(gateway.HandlerConfig{ Authenticator: gateway.NewAuthenticator(s), Registry: registry, - Heartbeat: s, PublicWSURL: publicWSURL, ResolveWSURL: resolve, + Heartbeat: s, PublicWSURL: publicWSURL, }) r := chi.NewRouter() r.Route("/api/v1", func(r chi.Router) { gateway.RegisterRoutes(r, h) }) diff --git a/services/agents-api/internal/store/runtime_adoption.go b/services/agents-api/internal/store/runtime_adoption.go deleted file mode 100644 index ab1c77705..000000000 --- a/services/agents-api/internal/store/runtime_adoption.go +++ /dev/null @@ -1,137 +0,0 @@ -package store - -import ( - "context" - "crypto/sha256" - "encoding/json" - "errors" - "fmt" - "reflect" - "time" - - "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/db/sqlc" - "github.com/jackc/pgx/v5/pgtype" -) - -var ErrRuntimeLegacyOwnership = errors.New("legacy sandbox ownership requires verification on the original backend") - -// RuntimeOwnershipVerifier is a startup-only, read-only check of retained resources. -// It must require positive allocation ownership evidence, never absence alone. -type RuntimeOwnershipVerifier func(context.Context, RuntimeAllocation) error - -// RuntimeOwnershipFailure carries only a sanitized startup diagnostic. -type RuntimeOwnershipFailure string - -func (e RuntimeOwnershipFailure) Error() string { return string(e) } -func runtimeOwnershipReason(err error) string { - var reason RuntimeOwnershipFailure - if errors.As(err, &reason) { - switch reason { - case "resource_missing", "ownership_mismatch", "resource_unconfirmed", "snapshot_unconfirmed", "provider_unavailable": - return string(reason) - } - } - return "provider_unavailable" -} - -type runtimeAdoptionPlan struct { - deployment sqlc.RuntimeDeployment - digest [32]byte -} - -func (s *Store) verifyLegacyRuntimeAdoption(ctx context.Context, selected *RuntimeDeployment, verify RuntimeOwnershipVerifier) (*runtimeAdoptionPlan, error) { - if selected == nil || selected.ProviderKind == "" { - return nil, nil - } - if err := s.CheckExecutionOwnership(ctx); err != nil { - return nil, err - } - read, cancel := context.WithTimeout(ctx, executionTransactionTimeout) - previous, err := s.queries.GetRuntimeDeployment(read) - cancel() - if err != nil { - return nil, err - } - if previous.ProviderKind != "" { - return nil, nil - } - // Configuration identity is a prerequisite, not evidence of a physical backend. - if runtimeUUID(previous.InstallationID) != selected.InstallationID || previous.BackendFingerprint != selected.BackendFingerprint { - return nil, nil - } - digest, _, err := legacyRuntimeDigest(ctx, s.queries, func(row sqlc.ListLegacyRuntimeAllocationsRow) error { - if selected.LocalNodeID == "" || verify == nil || runtimeUUID(row.RuntimeAllocation.ProviderKey) != selected.InstallationID { - return ErrRuntimeLegacyOwnership - } - if err := s.CheckExecutionOwnership(ctx); err != nil { - return err - } - allocation := runtimeAllocationFromRow(row.RuntimeAllocation, row.SessionID, row.TenantID, row.DeletedAt, false) - check, cancel := context.WithTimeout(ctx, 30*time.Second) - err := verify(check, allocation) - cancel() - if ownerErr := s.CheckExecutionOwnership(ctx); ownerErr != nil { - return ownerErr - } - if err != nil { - return fmt.Errorf("%w: allocation %s (%s); restore the original backend or resolve its resources with the previous Core before upgrading", ErrRuntimeLegacyOwnership, allocation.ID, runtimeOwnershipReason(err)) - } - return nil - }) - if err != nil { - return nil, err - } - return &runtimeAdoptionPlan{deployment: previous, digest: digest}, nil -} - -// Read in bounded pages; no external resource call holds a database transaction. -func legacyRuntimeDigest(ctx context.Context, q *sqlc.Queries, visit func(sqlc.ListLegacyRuntimeAllocationsRow) error) ([32]byte, int, error) { - h := sha256.New() - count := 0 - after := pgtype.UUID{Valid: true} - for { - read, cancel := context.WithTimeout(ctx, executionTransactionTimeout) - rows, err := q.ListLegacyRuntimeAllocations(read, after) - cancel() - if err != nil { - return [32]byte{}, 0, err - } - if len(rows) == 0 { - break - } - for _, row := range rows { - raw, err := json.Marshal(row) - if err != nil { - return [32]byte{}, 0, err - } - _, _ = h.Write(raw) - count++ - after = row.RuntimeAllocation.ID - if visit != nil { - if err := visit(row); err != nil { - return [32]byte{}, 0, err - } - } - } - } - var digest [32]byte - copy(digest[:], h.Sum(nil)) - return digest, count, nil -} - -func checkLegacyRuntimeAdoption(ctx context.Context, q *sqlc.Queries, previous sqlc.RuntimeDeployment, plan *runtimeAdoptionPlan) error { - digest, count, err := legacyRuntimeDigest(ctx, q, nil) - if err != nil { - return err - } - if plan == nil { - if count != 0 { - return ErrRuntimeLegacyOwnership - } - return nil - } - if !reflect.DeepEqual(previous, plan.deployment) || digest != plan.digest { - return fmt.Errorf("%w: retained receipts changed during verification; restart Core to verify again", ErrRuntimeLegacyOwnership) - } - return nil -} diff --git a/services/agents-api/internal/store/runtime_adoption_test.go b/services/agents-api/internal/store/runtime_adoption_test.go index 6e6c9a310..f1ec9bcfd 100644 --- a/services/agents-api/internal/store/runtime_adoption_test.go +++ b/services/agents-api/internal/store/runtime_adoption_test.go @@ -1,11 +1,9 @@ package store import ( - "context" - "errors" - "strings" + "fmt" + "reflect" "testing" - "time" "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/device" "github.com/google/uuid" @@ -40,132 +38,35 @@ func requireNoLegacyBinding(t *testing.T, s *Store) { t.Fatal("partial adoption", kind, nodes, placements, bound) } } -func TestRuntimeLegacyAdoptionRequiresEveryOwnershipReceipt(t *testing.T) { - for _, failure := range []string{"no_verifier", "resource_missing", "provider_unavailable", "ownership_mismatch", "snapshot_unconfirmed", "secret diagnostic"} { - t.Run(failure, func(t *testing.T) { - s, w, d, _ := legacyAdoptionFixture(t) - var verify RuntimeOwnershipVerifier - if failure != "no_verifier" { - verify = func(context.Context, RuntimeAllocation) error { return RuntimeOwnershipFailure(failure) } - } - err := w.ConfigureRuntimeDeployment(t.Context(), &d, verify) - if !errors.Is(err, ErrRuntimeLegacyOwnership) || strings.Contains(err.Error(), "secret diagnostic") { - t.Fatal(err) - } - if failure != "no_verifier" && failure != "secret diagnostic" && !strings.Contains(err.Error(), failure) { - t.Fatal("missing safe reason", err) - } - requireNoLegacyBinding(t, s) - }) - } -} -func TestRuntimeLegacyAdoptionRechecksCompletePlan(t *testing.T) { - for _, change := range []string{"compute_state", "state", "create_settled", "new_allocation", "lease_lost"} { - t.Run(change, func(t *testing.T) { - s, w, d, a := legacyAdoptionFixture(t) - verify := func(ctx context.Context, _ RuntimeAllocation) error { - switch change { - case "compute_state": - runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_allocations SET compute_state='{"changed":true}' WHERE id=$1`, a.ID) - case "state": - runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_allocations SET state='cleanup_pending' WHERE id=$1`, a.ID) - case "create_settled": - runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_allocations SET create_settled=true WHERE id=$1`, a.ID) - case "new_allocation": - _, e := localEnvironment(t, s, a.TenantID) - added, err := w.ReserveRuntimeAllocation(ctx, a.TenantID, e.ID, d.InstallationID, device.HashCredential("new")) +func TestHistoricalRuntimeResourcesCannotBeAdopted(t *testing.T) { + for _, state := range []string{"creating", "running", "cleanup_pending", "released"} { + for _, pending := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/pending=%t", state, pending), func(t *testing.T) { + s, w, d, a := legacyAdoptionFixture(t) + runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_allocations SET state=$2,create_settled=($2='released'),released_at=CASE WHEN $2='released' THEN clock_timestamp() ELSE NULL END WHERE id=$1`, a.ID, state) + if pending { + _, _ = localEnvironment(t, s, a.TenantID) + } + before, err := s.GetRuntimeAllocation(t.Context(), a.TenantID, a.EnvironmentID) + if err != nil { + t.Fatal(err) + } + err = w.ConfigureRuntimeDeployment(t.Context(), &d) + if state == "released" && !pending { if err != nil { - t.Fatal(err) + t.Fatal("released history blocked fresh configuration", err) } - runtimeSuspensionSQL(t, s.pool, "UPDATE runtime_allocations SET id='00000000-0000-0000-0000-000000000001' WHERE id=$1", added.ID) - case "lease_lost": - if err := w.executionLease.Close(ctx); err != nil { - t.Fatal(err) + } else { + if err == nil { + t.Fatal("historical resources adopted") } + requireNoLegacyBinding(t, s) } - return nil - } - if err := w.ConfigureRuntimeDeployment(t.Context(), &d, verify); err == nil { - t.Fatal("changed adoption plan accepted") - } - requireNoLegacyBinding(t, s) - }) - } -} -func TestRuntimeLegacyAdoptionChecksAllPagesAndRollsBack(t *testing.T) { - s, w, d, a := legacyAdoptionFixture(t) - for range 33 { - _, e := localEnvironment(t, s, a.TenantID) - if _, err := w.ReserveRuntimeAllocation(t.Context(), a.TenantID, e.ID, d.InstallationID, device.HashCredential(uuid.NewString())); err != nil { - t.Fatal(err) - } - } - calls := 0 - verify := func(context.Context, RuntimeAllocation) error { - calls++ - if calls == 34 { - return RuntimeOwnershipFailure("ownership_mismatch") - } - return nil - } - if err := w.ConfigureRuntimeDeployment(t.Context(), &d, verify); err == nil || calls != 34 { - t.Fatal(calls, err) - } - requireNoLegacyBinding(t, s) -} -func TestRuntimeLegacyAdoptionRetainsStatesAndIgnoresReleasedHistory(t *testing.T) { - for _, state := range []string{"creating", "running", "cleanup_pending", "released"} { - t.Run(state, func(t *testing.T) { - s, w, d, a := legacyAdoptionFixture(t) - runtimeSuspensionSQL(t, s.pool, `UPDATE runtime_allocations SET state=$2,create_settled=($2='released'), released_at=CASE WHEN $2='released' THEN clock_timestamp() ELSE NULL END WHERE id=$1`, a.ID, state) - pending, _ := localEnvironment(t, s, a.TenantID) - calls := 0 - verify := func(context.Context, RuntimeAllocation) error { calls++; return nil } - if err := w.ConfigureRuntimeDeployment(t.Context(), &d, verify); err != nil { - t.Fatal(err) - } - got, err := s.GetRuntimeAllocation(t.Context(), a.TenantID, a.EnvironmentID) - if err != nil || got.State != state || got.CreateSettled != (state == "released") { - t.Fatal(got, err) - } - if state == "released" { - if calls != 0 || got.NodeID != "" { - t.Fatal("released history was assigned", calls, got) - } - if _, err := sessionRuntimePlacement(t.Context(), s, a.TenantID, a.SessionID); !errors.Is(err, ErrNotFound) { - t.Fatal("released history placement fabricated", err) + after, err := s.GetRuntimeAllocation(t.Context(), a.TenantID, a.EnvironmentID) + if err != nil || !reflect.DeepEqual(before, after) { + t.Fatal("retained receipt changed", err) } - } else if calls != 1 || got.NodeID != d.LocalNodeID { - t.Fatal(calls, got) - } - if placement, err := sessionRuntimePlacement(t.Context(), s, a.TenantID, pending.ID); err != nil || placement.NodeID != d.LocalNodeID { - t.Fatal(placement, err) - } - if err := w.ConfigureRuntimeDeployment(t.Context(), &d, func(context.Context, RuntimeAllocation) error { t.Fatal("reverified fixed placement"); return nil }); err != nil { - t.Fatal(err) - } - }) - } -} -func TestRuntimeLegacyAdoptionFinalRowLockIsBounded(t *testing.T) { - s, w, d, a := legacyAdoptionFixture(t) - tx, err := s.pool.Begin(t.Context()) - if err != nil { - t.Fatal(err) - } - defer tx.Rollback(context.Background()) - verify := func(context.Context, RuntimeAllocation) error { - if _, err := tx.Exec(t.Context(), "SELECT id FROM runtime_allocations WHERE id=$1 FOR UPDATE", a.ID); err != nil { - t.Fatal(err) + }) } - return nil - } - started := time.Now() - if err := w.ConfigureRuntimeDeployment(t.Context(), &d, verify); err == nil { - t.Fatal("locked final receipt admitted") - } - if time.Since(started) > executionTransactionTimeout+2*time.Second { - t.Fatal("unbounded final lock wait") } - requireNoLegacyBinding(t, s) } diff --git a/services/agents-api/internal/store/runtime_deployment.go b/services/agents-api/internal/store/runtime_deployment.go index 84e69952a..f4a0e5ec1 100644 --- a/services/agents-api/internal/store/runtime_deployment.go +++ b/services/agents-api/internal/store/runtime_deployment.go @@ -26,7 +26,7 @@ type RuntimeDeployment struct { // ConfigureRuntimeDeployment runs before Worker startup under its execution lease. // AdmissionPaused must be committed for the old installation before any switch. // A nil selection never forgets the previous identity or unresolved resources. -func (s *Store) ConfigureRuntimeDeployment(ctx context.Context, selected *RuntimeDeployment, verify RuntimeOwnershipVerifier) error { +func (s *Store) ConfigureRuntimeDeployment(ctx context.Context, selected *RuntimeDeployment) error { if s.executionLease == nil { return ErrInvalidInput } @@ -46,10 +46,6 @@ func (s *Store) ConfigureRuntimeDeployment(ctx context.Context, selected *Runtim } update = sqlc.SetRuntimeDeploymentParams{InstallationID: id, BackendFingerprint: selected.BackendFingerprint, AdmissionPaused: selected.AdmissionPaused} } - plan, err := s.verifyLegacyRuntimeAdoption(ctx, selected, verify) - if err != nil { - return err - } ctx, cancel := context.WithTimeout(ctx, executionTransactionTimeout) defer cancel() return s.executionLease.transaction(ctx, func(tx pgx.Tx) error { @@ -65,7 +61,7 @@ func (s *Store) ConfigureRuntimeDeployment(ctx context.Context, selected *Runtim if err := q.SetRuntimeDeployment(ctx, update); err != nil { return err } - return configureRuntimeManager(ctx, q, previous, selected, plan) + return configureRuntimeManager(ctx, q, previous, selected) } resources, err := q.CountRuntimeDeploymentResources(ctx) if err != nil { @@ -92,7 +88,7 @@ func (s *Store) ConfigureRuntimeDeployment(ctx context.Context, selected *Runtim if err := q.SetRuntimeDeployment(ctx, update); err != nil { return err } - return configureRuntimeManager(ctx, q, previous, selected, plan) + return configureRuntimeManager(ctx, q, previous, selected) }) } @@ -112,7 +108,7 @@ func checkRuntimeDeploymentAdmission(ctx context.Context, q *sqlc.Queries, insta if current.AdmissionPaused { return fmt.Errorf("%w: sandbox creation is paused for provider maintenance", ErrEnvironmentUnavailable) } - if unspecifiedNodeDeployment(current) { + if _, err := deploymentSpecification(current); current.ProviderKind != "" && err != nil { return fmt.Errorf("%w: sandbox creation requires a deployment specification", ErrEnvironmentUnavailable) } if installation != "" { diff --git a/services/agents-api/internal/store/runtime_deployment_test.go b/services/agents-api/internal/store/runtime_deployment_test.go index 5d5392ee6..652d06121 100644 --- a/services/agents-api/internal/store/runtime_deployment_test.go +++ b/services/agents-api/internal/store/runtime_deployment_test.go @@ -19,7 +19,7 @@ func deploymentSelection() RuntimeDeployment { func deploymentConfigure(t *testing.T, w *Store, config *RuntimeDeployment) { t.Helper() - if err := w.ConfigureRuntimeDeployment(t.Context(), config, nil); err != nil { + if err := w.ConfigureRuntimeDeployment(t.Context(), config); err != nil { t.Fatal(err) } if config != nil && config.ProviderKind != "" { @@ -46,7 +46,7 @@ func TestRuntimeDeploymentRequiresMaintenanceBeforeIdentityChange(t *testing.T) w := executionLease(t, s).Store() old := deploymentSelection() deploymentConfigure(t, w, &old) - if err := s.ConfigureRuntimeDeployment(t.Context(), &old, nil); !errors.Is(err, ErrInvalidInput) { + if err := s.ConfigureRuntimeDeployment(t.Context(), &old); !errors.Is(err, ErrInvalidInput) { t.Fatal("unleased configuration accepted", err) } for _, changeID := range []bool{false, true} { @@ -57,7 +57,7 @@ func TestRuntimeDeploymentRequiresMaintenanceBeforeIdentityChange(t *testing.T) next.BackendFingerprint = strings.Repeat("b", 64) } next.AdmissionPaused = true - if err := w.ConfigureRuntimeDeployment(t.Context(), &next, nil); err == nil || !strings.Contains(err.Error(), "maintenance") { + if err := w.ConfigureRuntimeDeployment(t.Context(), &next); err == nil || !strings.Contains(err.Error(), "maintenance") { t.Fatal("identity changed before prior maintenance", err) } } @@ -66,7 +66,7 @@ func TestRuntimeDeploymentRequiresMaintenanceBeforeIdentityChange(t *testing.T) next := old next.BackendFingerprint = strings.Repeat("b", 64) next.AdmissionPaused = false - if err := w.ConfigureRuntimeDeployment(t.Context(), &next, nil); err == nil { + if err := w.ConfigureRuntimeDeployment(t.Context(), &next); err == nil { t.Fatal("switch reopened creation in same operation") } next.AdmissionPaused = true @@ -82,7 +82,7 @@ func TestRuntimeDeploymentRequiresMaintenanceBeforeIdentityChange(t *testing.T) deploymentConfigure(t, w, &next) another := deploymentSelection() another.AdmissionPaused = true - if err := w.ConfigureRuntimeDeployment(t.Context(), &another, nil); err == nil { + if err := w.ConfigureRuntimeDeployment(t.Context(), &another); err == nil { t.Fatal("nil selection erased the maintenance prerequisite") } } @@ -99,10 +99,10 @@ func TestRuntimeDeploymentPendingSessionsCannotMigrate(t *testing.T) { deploymentConfigure(t, w, &old) next := deploymentSelection() next.AdmissionPaused = true - if err := w.ConfigureRuntimeDeployment(t.Context(), &next, nil); err == nil || !strings.Contains(err.Error(), "1 pending hosted") { + if err := w.ConfigureRuntimeDeployment(t.Context(), &next); err == nil || !strings.Contains(err.Error(), "1 pending hosted") { t.Fatal("pending Session migrated", err) } - if err := w.ConfigureRuntimeDeployment(t.Context(), nil, nil); err == nil { + if err := w.ConfigureRuntimeDeployment(t.Context(), nil); err == nil { t.Fatal("pending Session orphaned by removing provider") } if err := s.DeleteSession(t.Context(), tenant, session.ID); err != nil { @@ -121,7 +121,7 @@ func TestRuntimeDeploymentUnknownAllocationsBlockAdoptionAndSwitch(t *testing.T) if err != nil { t.Fatal(err) } - if err := w.ConfigureRuntimeDeployment(t.Context(), &old, nil); err == nil || !strings.Contains(err.Error(), "no verified backend identity") { + if err := w.ConfigureRuntimeDeployment(t.Context(), &old); err == nil || !strings.Contains(err.Error(), "no verified backend identity") { t.Fatal("legacy allocation silently adopted", err) } if err := s.DeleteSession(t.Context(), tenant, session.ID); err != nil { @@ -133,7 +133,7 @@ func TestRuntimeDeploymentUnknownAllocationsBlockAdoptionAndSwitch(t *testing.T) if _, err := w.ReleaseRuntimeAllocation(t.Context(), owner); !errors.Is(err, ErrTurnConflict) { t.Fatal("unknown creation lost cleanup ownership", err) } - if err := w.ConfigureRuntimeDeployment(t.Context(), &old, nil); err == nil { + if err := w.ConfigureRuntimeDeployment(t.Context(), &old); err == nil { t.Fatal("deleted unknown allocation did not block adoption") } if _, err := w.SettleRuntimeCreation(t.Context(), owner); err != nil { @@ -152,10 +152,10 @@ func TestRuntimeDeploymentUnknownAllocationsBlockAdoptionAndSwitch(t *testing.T) deploymentConfigure(t, w, &old) next := deploymentSelection() next.AdmissionPaused = true - if err := w.ConfigureRuntimeDeployment(t.Context(), &next, nil); err == nil || !strings.Contains(err.Error(), "1 unreleased allocations") { + if err := w.ConfigureRuntimeDeployment(t.Context(), &next); err == nil || !strings.Contains(err.Error(), "1 unreleased allocations") { t.Fatal("unknown creation did not block switch", err) } - if err := w.ConfigureRuntimeDeployment(t.Context(), nil, nil); err == nil { + if err := w.ConfigureRuntimeDeployment(t.Context(), nil); err == nil { t.Fatal("removing adapter orphaned unknown creation") } replay, err := w.ReserveRuntimeAllocation(t.Context(), tenant, environment.ID, old.InstallationID, device.HashCredential(uuid.NewString())) @@ -282,7 +282,7 @@ func TestRuntimeDeploymentRetainedResourcesBlockSwitchWithoutMutation(t *testing } next := deploymentSelection() next.AdmissionPaused = true - if err := w.ConfigureRuntimeDeployment(t.Context(), &next, nil); err == nil || !strings.Contains(err.Error(), "1 unreleased allocations") { + if err := w.ConfigureRuntimeDeployment(t.Context(), &next); err == nil || !strings.Contains(err.Error(), "1 unreleased allocations") { t.Fatal("retained resource allowed switch", state, err) } if err := pool.QueryRow(t.Context(), "SELECT to_jsonb(a)::text FROM runtime_allocations a WHERE id=$1", owner.ID).Scan(&after); err != nil { @@ -325,7 +325,7 @@ func TestRuntimeDeploymentAllocationBeforeMaintenanceRetainsOwnership(t *testing maintaining := make(chan error, 1) maintenance := config maintenance.AdmissionPaused = true - go func() { maintaining <- w.ConfigureRuntimeDeployment(ctx, &maintenance, nil) }() + go func() { maintaining <- w.ConfigureRuntimeDeployment(ctx, &maintenance) }() if err := tx.Commit(ctx); err != nil { t.Fatal(err) } @@ -341,7 +341,7 @@ func TestRuntimeDeploymentAllocationBeforeMaintenanceRetainsOwnership(t *testing } next := deploymentSelection() next.AdmissionPaused = true - if err := w.ConfigureRuntimeDeployment(ctx, &next, nil); err == nil || !strings.Contains(err.Error(), "1 unreleased allocations") { + if err := w.ConfigureRuntimeDeployment(ctx, &next); err == nil || !strings.Contains(err.Error(), "1 unreleased allocations") { t.Fatal("earlier in-flight allocation omitted from switch guard", err) } } diff --git a/services/agents-api/internal/store/runtime_idle_clock_test.go b/services/agents-api/internal/store/runtime_idle_clock_test.go index 89945fd0f..1d5fa0b80 100644 --- a/services/agents-api/internal/store/runtime_idle_clock_test.go +++ b/services/agents-api/internal/store/runtime_idle_clock_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "reflect" "testing" "time" @@ -204,55 +205,19 @@ func TestRootCompletionRejectsNonpositiveSourceTime(t *testing.T) { } } -func TestManagedIdleClockLegacyAdoptionStartsIdleOnce(t *testing.T) { - for _, skew := range []time.Duration{-269 * time.Second, 269 * time.Second} { - t.Run(skew.String(), func(t *testing.T) { - s, _ := newManagedTestStore(t) - w := executionLease(t, s).Store() - d := deploymentSelection() - deploymentConfigure(t, w, &d) - tenant := uuid.NewString() - _, environment := localEnvironment(t, s, tenant) - owner, err := w.ReserveRuntimeAllocation(t.Context(), tenant, environment.ID, d.InstallationID, device.HashCredential("runtime")) - if err != nil { - t.Fatal(err) - } - owner, err = w.ObserveRuntimeRunning(t.Context(), owner) - if err != nil { - t.Fatal(err) - } - owner, err = w.SetRuntimeCompute(t.Context(), owner, "running", json.RawMessage(`{"instance":"original"}`), nil, 0) - if err != nil { - t.Fatal(err) - } - until := runtimeDatabaseTime(t, s).Add(time.Hour) - runtimeSuspensionSQL(t, s.pool, "UPDATE runtime_allocations SET compute_activity_at=clock_timestamp()-interval '10 minutes',compute_retained_until=$2 WHERE id=$1", owner.ID, until) - turn := runtimeSuspensionCompleted(t, s.pool, owner) - source := runtimeDatabaseTime(t, s).Add(skew).Truncate(time.Millisecond) - runtimeSuspensionSQL(t, s.pool, "UPDATE turns SET completed_at=$2 WHERE id=$1", turn, source) - d.ProviderKind = "microsandbox" - d.LocalNodeID = uuid.NewString() - d.LocalCredentialSHA256 = device.HashCredential("node") - d.LocalMaxActive, d.LocalMaxRetained = 1, 1 - before := runtimeDatabaseTime(t, s) - if err := w.ConfigureRuntimeDeployment(t.Context(), &d, func(context.Context, RuntimeAllocation) error { return nil }); err != nil { - t.Fatal(err) - } - after := runtimeDatabaseTime(t, s) - adopted, err := s.GetRuntimeAllocation(t.Context(), tenant, environment.ID) - if err != nil || adopted.ID != owner.ID || adopted.NodeID != d.LocalNodeID || adopted.ComputeRevision != owner.ComputeRevision || string(adopted.ComputeState) != string(owner.ComputeState) || adopted.ComputeRetainedUntil == nil || !adopted.ComputeRetainedUntil.Equal(until) { - t.Fatal("adoption changed retained compute", adopted, err) - } - deploymentConfigure(t, w, &d) - unchanged, err := s.GetRuntimeAllocation(t.Context(), tenant, environment.ID) - if err != nil || !unchanged.ComputeActivityAt.Equal(adopted.ComputeActivityAt) { - t.Fatal("restart reset adoption idle anchor", unchanged, err) - } - verifyManagedIdleClock(t, s, w, adopted, before, after) - read, err := s.GetTurn(t.Context(), tenant, owner.SessionID, turn) - if err != nil || !read.CompletedAt.Equal(source) { - t.Fatal("adoption changed native completion timestamp", read, err) - } - }) +func TestManagedIdleClockReconnectPreservesReceipts(t *testing.T) { + s, _, owner := managedIdleClockFixture(t) + turn := runtimeSuspensionCompleted(t, s.pool, owner) + before, err := s.GetRuntimeAllocation(t.Context(), owner.TenantID, owner.EnvironmentID) + if err != nil { + t.Fatal(err) + } + onlineManagerNode(t, s, owner.NodeID) + after, err := s.GetRuntimeAllocation(t.Context(), owner.TenantID, owner.EnvironmentID) + if err != nil || !reflect.DeepEqual(before, after) { + t.Fatal("reconnect changed compute receipt or idle clock", err) + } + if _, err := s.GetTurn(t.Context(), owner.TenantID, owner.SessionID, turn); err != nil { + t.Fatal(err) } } diff --git a/services/agents-api/internal/store/runtime_node_deployment.go b/services/agents-api/internal/store/runtime_node_deployment.go index 6eb8b275b..9783c290e 100644 --- a/services/agents-api/internal/store/runtime_node_deployment.go +++ b/services/agents-api/internal/store/runtime_node_deployment.go @@ -10,7 +10,7 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) -func configureRuntimeManager(ctx context.Context, q *sqlc.Queries, previous sqlc.RuntimeDeployment, selected *RuntimeDeployment, plan *runtimeAdoptionPlan) error { +func configureRuntimeManager(ctx context.Context, q *sqlc.Queries, previous sqlc.RuntimeDeployment, selected *RuntimeDeployment) error { if selected.ProviderKind == "" { return nil } @@ -22,22 +22,16 @@ func configureRuntimeManager(ctx context.Context, q *sqlc.Queries, previous sqlc return err } if previous.ProviderKind == "" { - if err := checkLegacyRuntimeAdoption(ctx, q, previous, plan); err != nil { + resources, err := q.CountRuntimeDeploymentResources(ctx) + if err != nil { return err } + if resources.Allocations != 0 || resources.Pending != 0 { + return fmt.Errorf("cannot adopt historical sandbox resources: keep the original Core responsible for retained resources and install this release separately") + } } var localNode pgtype.UUID - if selected.LocalNodeID == "" { - if previous.ProviderKind == "" { - resources, err := q.CountRuntimeDeploymentResources(ctx) - if err != nil { - return err - } - if resources.Allocations != 0 || resources.Pending != 0 { - return fmt.Errorf("cannot adopt existing local resources without their verified local node") - } - } - } else { + if selected.LocalNodeID != "" { id, err := parseConnectionGeneration(selected.LocalNodeID) if err != nil { return err @@ -70,17 +64,6 @@ func configureRuntimeManager(ctx context.Context, q *sqlc.Queries, previous sqlc if err != nil { return err } - if previous.ProviderKind == "" { - if err := q.AdoptRuntimePlacements(ctx, sqlc.AdoptRuntimePlacementsParams{NodeID: id, ProviderKey: installation}); err != nil { - return err - } - if err := q.AdoptPendingRuntimePlacements(ctx, id); err != nil { - return err - } - if err := q.BindLegacyRuntimeAllocations(ctx, sqlc.BindLegacyRuntimeAllocationsParams{NodeID: id, ProviderKey: installation}); err != nil { - return err - } - } } return q.SetRuntimeManagerDeployment(ctx, sqlc.SetRuntimeManagerDeploymentParams{ProviderKind: selected.ProviderKind, LocalNodeID: localNode}) } diff --git a/services/agents-api/internal/store/runtime_node_presence.go b/services/agents-api/internal/store/runtime_node_presence.go index 5f8e7636c..aa0c6a19d 100644 --- a/services/agents-api/internal/store/runtime_node_presence.go +++ b/services/agents-api/internal/store/runtime_node_presence.go @@ -75,10 +75,6 @@ func (s *Store) heartbeatRuntimeNode(ctx context.Context, nodeID, connectionID s return ErrRuntimeNodeCredential } if protocol == 1 { - // Unspecified legacy nodes keep cleanup connectivity without inventing a generation. - if n.DeploymentGeneration == 0 { - return nil - } state := "failed" if health.ProviderReady { state = "ready" diff --git a/services/agents-api/internal/store/runtime_nodes.go b/services/agents-api/internal/store/runtime_nodes.go index 99f6fdcb7..8567521a5 100644 --- a/services/agents-api/internal/store/runtime_nodes.go +++ b/services/agents-api/internal/store/runtime_nodes.go @@ -164,7 +164,10 @@ func (s *Store) CreateRuntimeEnrollment(ctx context.Context, capacity RuntimeNod if d.ResetClear.Valid { return ErrSandboxResetInProgress } - if d.Mode != "nodes" || d.AdmissionPaused || unspecifiedNodeDeployment(d) { + if d.Mode != "nodes" || d.AdmissionPaused { + return ErrSandboxDeploymentConflict + } + if _, err := deploymentSpecification(d); err != nil { return ErrSandboxDeploymentConflict } if err := q.CreateRuntimeEnrollment(ctx, sqlc.CreateRuntimeEnrollmentParams{ID: pgtype.UUID{Bytes: id, Valid: true}, TokenSha256: runtimeTokenDigest(result.Token), InstallationID: d.InstallationID, MaxActive: int32(capacity.MaxActive), MaxRetained: int32(capacity.MaxRetained)}); err != nil { @@ -264,12 +267,6 @@ func (s *Store) AuthenticateRuntimeNode(ctx context.Context, nodeID, credential if n.InstallationID != d.InstallationID || d.Mode != "nodes" { return RuntimeNodeIdentity{}, ErrRuntimeNodeCredential } - if unspecifiedNodeDeployment(d) { - if n.SpecificationDigest != "" || n.DeploymentGeneration != 0 { - return RuntimeNodeIdentity{}, ErrRuntimeSpecificationMismatch - } - return nodeIdentity(n, d.ProviderKind), nil - } if err := validateNodeEnrollmentIdentity(ctx, s.queries, d, n); err != nil { return RuntimeNodeIdentity{}, err } diff --git a/services/agents-api/internal/store/runtime_nodes_test.go b/services/agents-api/internal/store/runtime_nodes_test.go index 9e943435e..9361aa077 100644 --- a/services/agents-api/internal/store/runtime_nodes_test.go +++ b/services/agents-api/internal/store/runtime_nodes_test.go @@ -231,42 +231,20 @@ func TestRuntimeNodesEnrollmentAndEpoch(t *testing.T) { t.Fatal("removed node credential accepted", err) } } -func TestRuntimeNodesLegacyAdoptionAndRetention(t *testing.T) { - s, _ := newManagedTestStore(t) - w := executionLease(t, s).Store() - d := deploymentSelection() - deploymentConfigure(t, w, &d) +func TestRuntimeNodesRetention(t *testing.T) { + s, w, next := managerFixture(t, 2, 2) tenant := uuid.NewString() - first, environment := localEnvironment(t, s, tenant) - owner, err := w.ReserveRuntimeAllocation(t.Context(), tenant, environment.ID, d.InstallationID, device.HashCredential("runtime")) + first, err := s.CreateSession(t.Context(), tenant, managerSessionInput(uuid.NewString())) if err != nil { t.Fatal(err) } - pending, _ := localEnvironment(t, s, tenant) - next := d - next.ProviderKind = "docker" - next.LocalNodeID = uuid.NewString() - next.LocalCredentialSHA256 = device.HashCredential("node") - next.LocalMaxActive = 1 - next.LocalMaxRetained = 1 - wrong := next - wrong.BackendFingerprint = strings.Repeat("c", 64) - if err := w.ConfigureRuntimeDeployment(t.Context(), &wrong, nil); err == nil { - t.Fatal("adopted wrong backend") - } - if err := w.ConfigureRuntimeDeployment(t.Context(), &next, func(context.Context, RuntimeAllocation) error { return nil }); err != nil { + retained, err := w.ReserveRuntimeAllocation(t.Context(), tenant, first.Environment.ID, next.InstallationID, device.HashCredential("runtime")) + if err != nil { t.Fatal(err) } - legacyRuntimeSpecification(t, w, next.ProviderKind) - retained, err := s.GetRuntimeAllocation(t.Context(), tenant, environment.ID) - if err != nil || retained.NodeID != next.LocalNodeID || retained.ProviderKey != d.InstallationID || retained.ID != owner.ID { - t.Fatal("adoption lost identity", retained, err) - } - for _, session := range []Session{first, pending} { - placement, err := sessionRuntimePlacement(t.Context(), s, tenant, session.ID) - if err != nil || placement.NodeID != next.LocalNodeID { - t.Fatal("legacy Session reassigned", placement, err) - } + pending, err := s.CreateSession(t.Context(), tenant, managerSessionInput(uuid.NewString())) + if err != nil { + t.Fatal(err) } if err := s.RemoveRuntimeNode(t.Context(), next.LocalNodeID); !errors.Is(err, ErrRuntimeNodeInUse) { t.Fatal(err) @@ -294,7 +272,7 @@ func TestRuntimeNodesLegacyAdoptionAndRetention(t *testing.T) { if err := s.RemoveRuntimeNode(t.Context(), next.LocalNodeID); !errors.Is(err, ErrRuntimeLocalNodeConfigured) { t.Fatal("configured local node was removed", err) } - if _, err := s.AuthenticateRuntimeNode(t.Context(), next.LocalNodeID, "node"); err != nil { + if _, err := s.AuthenticateRuntimeNode(t.Context(), next.LocalNodeID, "local-node-credential"); err != nil { t.Fatal("rejected removal changed local credentials", err) } next.AdmissionPaused = true @@ -397,7 +375,7 @@ func TestRuntimeNodesLongOfflineRetainsExactAllocation(t *testing.T) { } changed := d changed.LocalNodeID = uuid.NewString() - if err := w.ConfigureRuntimeDeployment(t.Context(), &changed, nil); err == nil { + if err := w.ConfigureRuntimeDeployment(t.Context(), &changed); err == nil { t.Fatal("lost local state created replacement identity") } onlineManagerNode(t, s, d.LocalNodeID) diff --git a/services/agents-api/internal/store/sandbox_deployment_mutations.go b/services/agents-api/internal/store/sandbox_deployment_mutations.go index 2d60b8729..202b5de40 100644 --- a/services/agents-api/internal/store/sandbox_deployment_mutations.go +++ b/services/agents-api/internal/store/sandbox_deployment_mutations.go @@ -251,7 +251,10 @@ func checkSandboxSwitch(ctx context.Context, q *sqlc.Queries, d sqlc.RuntimeDepl if d.ProviderKind == "" { return ErrSandboxNotConfigured } - if d.ProviderKind != input.Provider || unspecifiedNodeDeployment(d) { + if _, err := deploymentSpecification(d); err != nil { + return err + } + if d.ProviderKind != input.Provider { return &SandboxResetRequiredError{CurrentProvider: d.ProviderKind, RequestedProvider: input.Provider} } @@ -322,7 +325,12 @@ func (s *Store) CheckSandboxDeploymentSetup(ctx context.Context, installation st if err := validateSandboxSelection(input); err != nil { return err } - if d.ProviderKind != "" && (d.ProviderKind != input.Provider || unspecifiedNodeDeployment(d)) { + if d.ProviderKind != "" { + if _, err := deploymentSpecification(d); err != nil { + return err + } + } + if d.ProviderKind != "" && d.ProviderKind != input.Provider { return &SandboxResetRequiredError{CurrentProvider: d.ProviderKind, RequestedProvider: input.Provider} } return nil diff --git a/services/agents-api/internal/store/sandbox_deployment_setup.go b/services/agents-api/internal/store/sandbox_deployment_setup.go index 74aa89ce1..9b7a58a9c 100644 --- a/services/agents-api/internal/store/sandbox_deployment_setup.go +++ b/services/agents-api/internal/store/sandbox_deployment_setup.go @@ -55,7 +55,7 @@ func (s *Store) sandboxSetup(d sqlc.RuntimeDeployment) (SandboxSetup, error) { if err := json.Unmarshal(d.Specification, &result.Specification); err != nil { return SandboxSetup{}, err } - if d.ProviderKind != "" && !unspecifiedNodeDeployment(d) { + if d.ProviderKind != "" { if err := result.Specification.Validate(d.ProviderKind); err != nil { return SandboxSetup{}, err } @@ -102,6 +102,11 @@ func (s *Store) ClaimWebSandboxDeployment(ctx context.Context, installationID st return ErrSandboxDeploymentConflict } } + if d.ProviderKind != "" { + if _, err := deploymentSpecification(d); err != nil { + return err + } + } return q.ClaimWebSandboxDeployment(ctx, id) }) } diff --git a/services/agents-api/internal/store/sandbox_deployment_switch_test.go b/services/agents-api/internal/store/sandbox_deployment_switch_test.go index e17bb2e29..93615b8ce 100644 --- a/services/agents-api/internal/store/sandbox_deployment_switch_test.go +++ b/services/agents-api/internal/store/sandbox_deployment_switch_test.go @@ -335,7 +335,7 @@ func TestSandboxSwitchPreservesReleasedAllocationAndItemHistory(t *testing.T) { // with an empty specification and its nodes with empty digests. The retained // node reconnects to drain resources; fresh admission and node configuration // stay closed until an administrator replaces the selection. -func TestUnspecifiedNodeDeploymentDrainsBeforeReplacement(t *testing.T) { +func TestUnspecifiedNodeDeploymentRejectedWithoutMutation(t *testing.T) { _, pool := newManagedTestStore(t) cipher, _ := credentialcrypto.New(bytes.Repeat([]byte{7}, 32)) s := NewWithCredentialCipher(pool, cipher) @@ -364,12 +364,11 @@ func TestUnspecifiedNodeDeploymentDrainsBeforeReplacement(t *testing.T) { t.Fatal(err) } - if setup, err := s.GetSandboxSetup(t.Context()); err != nil || setup.Provider != "docker" { - t.Fatal("unspecified deployment could not load for draining", err) + if _, err := s.GetSandboxSetup(t.Context()); err == nil { + t.Fatal("missing deployment specification accepted") } - identity, err := s.AuthenticateRuntimeNode(t.Context(), node.NodeID, node.Credential) - if err != nil || identity.SpecificationDigest != "" || identity.DeploymentGeneration != 0 { - t.Fatal("retained unspecified node could not reconnect", identity, err) + if _, err := s.AuthenticateRuntimeNode(t.Context(), node.NodeID, node.Credential); !errors.Is(err, ErrRuntimeSpecificationMismatch) { + t.Fatal("unspecified node authenticated", err) } if _, err := s.RuntimeNodeConfiguration(t.Context(), node.NodeID, node.Credential); !errors.Is(err, ErrRuntimeSpecificationMismatch) { t.Fatal("node configuration served without a specification", err) @@ -381,14 +380,15 @@ func TestUnspecifiedNodeDeploymentDrainsBeforeReplacement(t *testing.T) { t.Fatal("unspecified deployment issued an enrollment token", err) } - if _, err := w.StartSandboxReset(SandboxResetTestContext(t.Context()), id, SandboxResetRequest{Clear: "auto", ExpectedGeneration: 1}); err != nil { - t.Fatal(err) + epoch := managerEpoch(t, s) + if err := w.ClaimWebSandboxDeployment(t.Context(), id); err == nil { + t.Fatal("unsupported installation claimed") } - view, err := resetAndSelect(t, w, id, 1, selection) - if err != nil || view.Generation != 3 || view.Specification == nil { - t.Fatal("replacement did not record the specification", view, err) + if managerEpoch(t, s) != epoch { + t.Fatal("refused startup changed owner epoch") } - if _, err := s.AuthenticateRuntimeNode(t.Context(), node.NodeID, node.Credential); !errors.Is(err, ErrRuntimeNodeCredential) { - t.Fatal("unspecified node survived replacement", err) + var specification string + if err := pool.QueryRow(t.Context(), "SELECT specification::text FROM runtime_deployment").Scan(&specification); err != nil || specification != "{}" { + t.Fatal("deployment silently repaired", specification, err) } } diff --git a/services/agents-api/internal/store/sandbox_specification.go b/services/agents-api/internal/store/sandbox_specification.go index cb39081a9..67a9c44b6 100644 --- a/services/agents-api/internal/store/sandbox_specification.go +++ b/services/agents-api/internal/store/sandbox_specification.go @@ -50,14 +50,6 @@ func SandboxSetupForSelection(installationID string, input SandboxDeploymentSetu return result, nil } -// unspecifiedNodeDeployment identifies a node-backed selection saved before -// deployments carried a specification; migration left the empty default. Its -// retained nodes may reconnect to drain resources, but it cannot create -// sandboxes or enroll nodes until an administrator replaces the selection. -func unspecifiedNodeDeployment(d sqlc.RuntimeDeployment) bool { - return d.WebManaged && d.Mode == "nodes" && (d.ProviderKind == "docker" || d.ProviderKind == "microsandbox") && string(d.Specification) == "{}" -} - func deploymentSpecification(d sqlc.RuntimeDeployment) (sandbox.DeploymentSpec, error) { var spec sandbox.DeploymentSpec if json.Unmarshal(d.Specification, &spec) != nil || spec.Validate(d.ProviderKind) != nil { diff --git a/services/agents-api/internal/store/sandbox_specification_store_test.go b/services/agents-api/internal/store/sandbox_specification_store_test.go index c6c370246..f118393fa 100644 --- a/services/agents-api/internal/store/sandbox_specification_store_test.go +++ b/services/agents-api/internal/store/sandbox_specification_store_test.go @@ -72,7 +72,7 @@ func TestSandboxSpecificationRoundTripAndFileConfigurationCannotOverride(t *test } file := RuntimeDeployment{InstallationID: view.InstallationID, BackendFingerprint: setup.BackendFingerprint, ProviderKind: provider, AdmissionPaused: true} for _, candidate := range []*RuntimeDeployment{nil, &file} { - if err := w.ConfigureRuntimeDeployment(t.Context(), candidate, nil); !errors.Is(err, ErrSandboxDeploymentConflict) { + if err := w.ConfigureRuntimeDeployment(t.Context(), candidate); !errors.Is(err, ErrSandboxDeploymentConflict) { t.Fatal("file configuration replaced database ownership", err) } }