diff --git a/integration/agent_run_node_unknown_test.go b/integration/agent_run_node_unknown_test.go new file mode 100644 index 000000000..052a3adf9 --- /dev/null +++ b/integration/agent_run_node_unknown_test.go @@ -0,0 +1,233 @@ +package integration + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/wanglongan587/cloud/internal/controlpb" +) + +// These cases use real Controller reports to distinguish a disconnected Node from one that has +// been unknown for the configured window. Database time is aged by a wide margin; no case sleeps. +func TestNodeUnknownDisconnectedWithinWindowKeepsDeliveryAndRelease(t *testing.T) { + for _, phase := range []string{"delivering", "releasing"} { + t.Run(phase, func(t *testing.T) { + f := setup(t) + f.store.DeliveryUnreachableAfter = time.Hour + scene := nodeUnknownScene(t, f, phase) + version := int64(f.scalar("SELECT version FROM node_instances WHERE id=$1", scene.nodeID)) + version = reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_CONNECTED) + must(t, f.ageRunWorkspaceNodeBy(scene.runID, 15*time.Minute)) + reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_DISCONNECTED) + runVersion := f.runVersion(scene.runID) + + passNodeUnknownWindow(t, f, phase) + + assertNodeUnknownStillWaiting(t, f, scene.runID, phase, runVersion) + }) + } +} + +func TestNodeUnknownDisconnectedPastWindowSettlesDeliveryAndRelease(t *testing.T) { + for _, phase := range []string{"delivering", "releasing"} { + t.Run(phase, func(t *testing.T) { + f := setup(t) + f.store.DeliveryUnreachableAfter = time.Hour + scene := nodeUnknownScene(t, f, phase) + version := int64(f.scalar("SELECT version FROM node_instances WHERE id=$1", scene.nodeID)) + version = reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_CONNECTED) + must(t, f.ageRunWorkspaceNodeBy(scene.runID, 2*time.Hour)) + reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_DISCONNECTED) + + passNodeUnknownWindow(t, f, phase) + + assertNodeUnknownExpired(t, f, scene.runID, phase) + }) + } +} + +func TestNodeUnknownReconnectRestartsWindowButRepeatedDisconnectDoesNot(t *testing.T) { + for _, phase := range []string{"delivering", "releasing"} { + t.Run(phase, func(t *testing.T) { + f := setup(t) + f.store.DeliveryUnreachableAfter = time.Hour + scene := nodeUnknownScene(t, f, phase) + version := int64(f.scalar("SELECT version FROM node_instances WHERE id=$1", scene.nodeID)) + version = reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_CONNECTED) + must(t, f.ageRunWorkspaceNodeBy(scene.runID, 15*time.Minute)) + previous := nodeUnknownLastSeen(t, f, scene.nodeID) + version = reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_DISCONNECTED) + if got := nodeUnknownLastSeen(t, f, scene.nodeID); !got.Equal(previous) { + t.Fatalf("disconnect must preserve the last connected instant: %s -> %s", previous, got) + } + + version = reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_CONNECTED) + reconnected := nodeUnknownLastSeen(t, f, scene.nodeID) + if !reconnected.After(previous) { + t.Fatalf("reconnect must restart the window on database time: %s -> %s", previous, reconnected) + } + version = reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_DISCONNECTED) + if got := nodeUnknownLastSeen(t, f, scene.nodeID); !got.Equal(reconnected) { + t.Fatalf("a new disconnect must retain the reconnected instant: %s -> %s", reconnected, got) + } + runVersion := f.runVersion(scene.runID) + passNodeUnknownWindow(t, f, phase) + assertNodeUnknownStillWaiting(t, f, scene.runID, phase, runVersion) + + // Expire the new window, then deliver another fresh disconnect report. Receiving that + // report is not evidence that Cloud can reach the Node and must not postpone give-up. + must(t, f.ageRunWorkspaceNodeBy(scene.runID, 2*time.Hour)) + expired := nodeUnknownLastSeen(t, f, scene.nodeID) + reportNodeUnknownConnection(t, f, scene.nodeID, version, controlpb.NodeConnection_NODE_CONNECTION_DISCONNECTED) + if got := nodeUnknownLastSeen(t, f, scene.nodeID); !got.Equal(expired) { + t.Fatalf("repeated disconnect must not refresh last_seen_at: %s -> %s", expired, got) + } + passNodeUnknownWindow(t, f, phase) + assertNodeUnknownExpired(t, f, scene.runID, phase) + }) + } +} + +// A terminated sandbox is immediately unknown to delivery, but release must still finish its +// Node-independent cleanup. D8 must not terminalize that real delete while it is at cleanup. +func TestNodeUnknownNoLiveSandboxReleasesDeliveryButPreservesCleanup(t *testing.T) { + f := setup(t) + f.store.DeliveryUnreachableAfter = time.Hour + scene := deliveringScene(t, f) + _, err := f.store.Pool.Exec(` + WITH ended AS ( + UPDATE sandbox_instances SET observed_state='terminated',terminated_at=clock_timestamp() + WHERE workspace_id=(SELECT workspace_id FROM issue_runs WHERE id=$1) AND terminated_at IS NULL + RETURNING id + ) + UPDATE node_instances SET connection_state='ended',ended_at=clock_timestamp(),version=version+1 + WHERE sandbox_instance_id IN (SELECT id FROM ended) AND ended_at IS NULL`, scene.runID) + must(t, err) + + // No heartbeat is aged: absence of a live sandbox is the immediate D5 condition. + must(t, f.store.GiveUpStaleDeliveriesOnce(context.Background())) + assertNodeUnknownExpired(t, f, scene.runID, "delivering") + c := &grpcController{ + t: t, f: f, ctx: asController("ctrl-a"), epoch: 1, + operations: controlpb.NewWorkspaceOperationServiceClient(f.controlConn), + } + op := c.claim().GetOperation() + op, err = c.advance(op) + must(t, err) + op, err = c.advance(op) + must(t, err) + if op.GetStep() != controlpb.OperationStep_OPERATION_STEP_CLEANUP || op.GetState() != controlpb.OperationState_OPERATION_STATE_RUNNING { + t.Fatalf("the real delete must reach cleanup without a Node, got %v", op) + } + version := f.runVersion(scene.runID) + + must(t, f.store.GiveUpStaleWorkspaceReleasesOnce(context.Background())) + + if got := f.runPhase(scene.runID); got != "releasing" { + t.Fatalf("D8 must leave Node-independent cleanup releasing, got %q", got) + } + if got := f.runVersion(scene.runID); got != version { + t.Fatalf("D8 must not rewrite a cleanup run: version %d -> %d", version, got) + } + if got := f.runFailureReason(scene.runID); got != "" { + t.Fatalf("cleanup must not acquire a give-up reason, got %q", got) + } + var step, state, code string + var opVersion int64 + must(t, f.store.Pool.QueryRow(`SELECT step,state,COALESCE(error_code,''),version FROM operations WHERE id=$1`, op.GetId()).Scan(&step, &state, &code, &opVersion)) + if step != "cleanup" || state != "running" || code != "" || opVersion != op.GetVersion() { + t.Fatalf("D8 must preserve the running cleanup: step=%s state=%s error=%s version=%d, want %d", step, state, code, opVersion, op.GetVersion()) + } +} + +// nodeUnknownScene reaches each tested phase through the real session/delivery path and gives its +// seeded sandbox the external identity required by the real Controller report route. +func nodeUnknownScene(t *testing.T, f *fixture, phase string) liveThreadScene { + t.Helper() + var scene liveThreadScene + switch phase { + case "delivering": + scene = deliveringScene(t, f) + case "releasing": + scene = driveToReleasing(t, f, controlpb.AgentSessionEndReason_AGENT_SESSION_END_REASON_USER_ENDED).liveThreadScene + default: + t.Fatalf("unsupported scene phase %q", phase) + } + // driveToReleasing uses continuous failure to reach its starting point. Only the unknown-Node + // window may decide any subsequent pass in these tests. + f.store.DeliveryGiveUpAfter = 0 + f.bindRunWorkspaceSandbox(t, scene.runID) + return scene +} + +func reportNodeUnknownConnection(t *testing.T, f *fixture, nodeID string, version int64, connection controlpb.NodeConnection) int64 { + t.Helper() + out, err := controlpb.NewNodeReportServiceClient(f.controlConn).ReportNodeStatus(asController("ctrl-a"), &controlpb.ReportNodeStatusRequest{ + SubmissionId: uuid.NewString(), Epoch: 1, NodeInstanceId: nodeID, + Version: version, Connection: connection, Initialized: true, + }) + must(t, err) + return out.GetNode().GetVersion() +} + +func nodeUnknownLastSeen(t *testing.T, f *fixture, nodeID string) time.Time { + t.Helper() + var seen time.Time + must(t, f.store.Pool.QueryRow(`SELECT last_seen_at FROM node_instances WHERE id=$1`, nodeID).Scan(&seen)) + return seen +} + +func passNodeUnknownWindow(t *testing.T, f *fixture, phase string) { + t.Helper() + if phase == "delivering" { + must(t, f.store.GiveUpStaleDeliveriesOnce(context.Background())) + return + } + must(t, f.store.GiveUpStaleWorkspaceReleasesOnce(context.Background())) +} + +func assertNodeUnknownStillWaiting(t *testing.T, f *fixture, runID, phase string, version int64) { + t.Helper() + if got := f.runPhase(runID); got != phase { + t.Fatalf("a disconnect within the window must stay %s, got %q", phase, got) + } + if got := f.runVersion(runID); got != version { + t.Fatalf("waiting must not rewrite the run: version %d -> %d", version, got) + } + if got := f.runFailureReason(runID); got != "" { + t.Fatalf("waiting must not record give-up, got %q", got) + } + ops := deleteOperations(f.runOperations(runID)) + if phase == "delivering" { + if len(ops) != 0 { + t.Fatalf("waiting delivery must not declare deletion, got %v", ops) + } + return + } + if len(ops) != 1 || ops[0].S("state") != "queued" || ops[0].S("errorCode") != "" { + t.Fatalf("waiting release must preserve its queued delete without failure, got %v", ops) + } +} + +func assertNodeUnknownExpired(t *testing.T, f *fixture, runID, phase string) { + t.Helper() + if phase == "releasing" { + assertGiveUpSettlement(t, f, runID, "completed") + return + } + if got := f.runPhase(runID); got != "releasing" { + t.Fatalf("D5 must release an unknown Node's delivery, got %q", got) + } + if got := f.runResult(runID).S("deliveryState"); got != "failed" { + t.Fatalf("D5 must record deliveryState=failed, got %q", got) + } + if got, _ := f.runStatusVersion(runID); got != "completed" { + t.Fatalf("D5 must retain the session's completed outcome, got %q", got) + } + if ops := deleteOperations(f.runOperations(runID)); len(ops) != 1 || ops[0].S("state") != "queued" { + t.Fatalf("D5 must declare a queued delete, got %v", ops) + } +} diff --git a/internal/core/agent_run_release.go b/internal/core/agent_run_release.go index 596425224..3a93e5428 100644 --- a/internal/core/agent_run_release.go +++ b/internal/core/agent_run_release.go @@ -133,7 +133,7 @@ func (s *Store) releaseAfterDelivery(t *transaction, o Object, state, revisionID // since the run's FIRST delivery work item was declared. While the run is // `delivering` every attempt has failed — a success would have left the phase — // so that instant is the start of the unbroken failure run D5 measures. -// unreachable Workspace: the run's Workspace has no live sandbox carrying a connected Node seen +// unreachable Workspace: the run's Workspace has no live sandbox carrying a Node last reachable // within `DeliveryUnreachableAfter` (D5's default 30 min). A Workspace whose // Node's state has been unknown that long cannot deliver, so waiting out the // full failure window would only hold the sandbox longer. @@ -156,10 +156,11 @@ func (s *Store) deliveryGivenUp(t *transaction, o Object) bool { // runWorkspaceNodeUnknown is the one authoritative judgement that a run Workspace's Node state is // unknown, and the only place the heartbeat evidence behind it is read: the Workspace is bound to this -// run and no live sandbox of it carries a Node that is connected and was last seen inside window. +// run and no live sandbox of it carries a Node last confirmed reachable inside window. // IssueRun D5's delivery give-up, IssueRun D8's delete give-up and the workflow scheduler's in-flight -// session settlement (workflowAgentSessionOutcome) all ask this one function, so none of them can -// drift into a second definition of "the Node is unreachable". +// session settlement (workflowAgentSessionOutcome) and cancellation (advanceCancelledWorkflowRun) +// all ask this one function, so none of them can drift into a second definition of +// "the Node is unreachable". // // The binding is whichever column ties the Workspace to its run — `issue_run_id` for an IssueRun, // `workflow_run_id` for a WorkflowRun. The two are mutually exclusive (0039's XOR), so one comparison @@ -188,7 +189,7 @@ func runWorkspaceNodeUnknown(t *transaction, runID, wid string, window time.Dura SELECT 1 FROM sandbox_instances sb JOIN node_instances n ON n.sandbox_instance_id = sb.id WHERE sb.workspace_id = w.id AND sb.terminated_at IS NULL - AND n.ended_at IS NULL AND n.connection_state='connected' + AND n.ended_at IS NULL AND n.last_seen_at > clock_timestamp() - make_interval(secs => $3))`, wid, runID, window.Seconds()) != nil } diff --git a/internal/core/node.go b/internal/core/node.go index f2f81f952..d186817ef 100644 --- a/internal/core/node.go +++ b/internal/core/node.go @@ -98,15 +98,17 @@ func ensuredNode(t *transaction, sid, nodeID string) { require(e != nil && e.O("result").S("nodeId") == nodeID, 409, "node_identity_mismatch") } -// nodeStatus records one connection report and refreshes the heartbeat. A disconnect closes -// admission on a ready Workspace; any report also drops earlier idle evidence, which must be -// given again for the admission epoch it applies to. +// nodeStatus records one connection report and refreshes the heartbeat only while connected. +// A disconnect closes admission on a ready Workspace; any report also drops earlier idle evidence, +// which must be given again for the admission epoch it applies to. func nodeStatus(t *transaction, r *ControlRequest, n, w Object) { version(n, r.Body.N("version")) state := r.Body.S("connectionState") require(state == "connected" || state == "disconnected", 400, "invalid_node_state") require(!n.B("initialized") || r.Body.B("initialized"), 409, "initialization_regression") - t.exec("UPDATE node_instances SET connection_state=$2,initialized=$3,last_seen_at=clock_timestamp(),idle_admission_epoch=NULL,version=version+1 WHERE id=$1", n.S("id"), state, r.Body.B("initialized")) + // Keep the last confirmed reachable instant: repeated disconnect reports must not restart + // D5/D8's unreachable window, while a connected report naturally starts a new window. + t.exec("UPDATE node_instances SET connection_state=$2,initialized=$3,last_seen_at=CASE WHEN $2='connected' THEN clock_timestamp() ELSE last_seen_at END,idle_admission_epoch=NULL,version=version+1 WHERE id=$1", n.S("id"), state, r.Body.B("initialized")) if state == "disconnected" { closeUnavailable(t, w) } else if r.Body.B("initialized") && w.S("desiredState") == "running" && w.S("observedState") == "unavailable" && w.S("baseCommitId") != "" { diff --git a/internal/core/workflow_run_session_db_test.go b/internal/core/workflow_run_session_db_test.go index 16bce9df6..cebdf13e2 100644 --- a/internal/core/workflow_run_session_db_test.go +++ b/internal/core/workflow_run_session_db_test.go @@ -933,6 +933,114 @@ func TestWorkflowSessionNodeSilenceWithinWindowHolds(t *testing.T) { } } +// An explicit disconnect is not an expired unknown-Node window. The real connection-report writer +// preserves the last connected instant, and the scheduler holds the session until that instant ages +// past the configured limit. +func TestWorkflowSessionDisconnectedWithinWindowHoldsUntilExpired(t *testing.T) { + store := dispatcherDB(t) + store.DeliveryUnreachableAfter = time.Hour + runID, _, _, _, _ := seedWorkflowSessionScene(t, store) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance 1: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + dispatchSession(t, store, runID, "a", nodeService, spec) + if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-interval '5 minutes' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatalf("age the last connected report inside the window: %v", err) + } + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + n := tx.one("SELECT * FROM node_instances WHERE workspace_id=$1 AND node_id=$2", wsID, nodeService) + w := tx.one("SELECT * FROM workspaces WHERE id=$1", wsID) + nodeStatus(tx, &ControlRequest{Body: Object{ + "version": n.N("version"), "connectionState": "disconnected", "initialized": true, + }}, n, w) + return Object{} + }); err != nil { + t.Fatalf("report the Node disconnected: %v", err) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance after disconnect within the window: %v", err) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + agent, _ := asWorkflowObject(states["a"]) + if status != "running" || runErr != "" || agent == nil || agent.S("status") != "running" { + t.Fatalf("a disconnect within the window settled run %q error %q states=%v; want it held", status, runErr, states) + } + + if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-interval '2 hours' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatalf("age the disconnected Node past the window: %v", err) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance after the disconnected window expires: %v", err) + } + status, runErr, states, _ = readWorkflowRun(t, store, runID) + agent, _ = asWorkflowObject(states["a"]) + if status != "failed" || runErr != workflowSessionNodeUnavailable || agent == nil || agent.S("status") != "failed" || agent.S("error") != workflowSessionNodeUnavailable { + t.Fatalf("past the disconnected window: run %q error %q states=%v; want failed/%s", status, runErr, states, workflowSessionNodeUnavailable) + } + if reason := agent.O("output").S("reason"); reason != workflowSessionNodeUnavailable { + t.Fatalf("node a output = %v; want reason %s", agent.O("output"), workflowSessionNodeUnavailable) + } +} + +// Cancellation inside the disconnected window still issues EndSession and waits for its own +// confirmation deadline. Aging that command alone must settle cancellation without waiting for +// the longer unknown-Node window or changing the production confirmation timeout. +func TestWorkflowRunCancelDisconnectedWithinWindowWaitsForEndConfirmation(t *testing.T) { + store := dispatcherDB(t) + store.DeliveryUnreachableAfter = time.Hour + runID, tid, _, _, _ := seedWorkflowSessionScene(t, store) + wid := workflowRunWorkflowID(t, store, runID) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance 1: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + execution := dispatchSession(t, store, runID, "a", nodeService, spec) + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + n := tx.one("SELECT * FROM node_instances WHERE workspace_id=$1 AND node_id=$2", wsID, nodeService) + w := tx.one("SELECT * FROM workspaces WHERE id=$1", wsID) + nodeStatus(tx, &ControlRequest{Body: Object{ + "version": n.N("version"), "connectionState": "disconnected", "initialized": true, + }}, n, w) + return Object{} + }); err != nil { + t.Fatalf("report the Node disconnected: %v", err) + } + mustCancelWorkflowRun(t, store, tid, wid, runID, "wf-cancel-disconnected-window", cancelCaller(t, store, tid)) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance cancellation within the disconnected window: %v", err) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + agent, _ := asWorkflowObject(states["a"]) + if status != "running" || runErr != "" || agent == nil || agent.S("status") != "running" { + t.Fatalf("cancellation skipped confirmation: run %q error %q states=%v; want it held", status, runErr, states) + } + commands := claimCommands(t, store) + if len(commands) != 1 { + t.Fatalf("released %d commands; want one end request: %v", len(commands), commands) + } + cmd := commands[0] + if cmd.S("kind") != "end_session" || cmd.O("body").S("reason") != "cancelled" || cmd.S("runId") != runID || cmd.S("executionId") != execution || cmd.S("nodeId") != nodeService { + t.Fatalf("command = %v; want end_session/cancelled for run %s execution %s node %s", cmd, runID, execution, nodeService) + } + + // Only the command ages: the disconnected Node remains inside the one-hour window, while the + // existing five-minute EndSession confirmation deadline has elapsed on database time. + if _, err := store.Pool.Exec(`UPDATE thread_commands SET created_at=clock_timestamp()-interval '6 minutes' WHERE id=$1`, cmd.S("commandId")); err != nil { + t.Fatalf("age the end request past its confirmation deadline: %v", err) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance past EndSession confirmation: %v", err) + } + status, runErr, states, _ = readWorkflowRun(t, store, runID) + agent, _ = asWorkflowObject(states["a"]) + if status != "cancelled" || runErr != "" || agent == nil || agent.S("status") != "cancelled" || agent.S("error") != "" || agent.O("output").S("reason") != "cancelled" { + t.Fatalf("after EndSession confirmation expires: run %q error %q states=%v; want cancelled with no error", status, runErr, states) + } +} + // An unconfigured window fails closed here exactly as it does in the IssueRun release paths: with no // configured tolerance for a silent Node, Cloud holds the session rather than failing a node on a // limit nobody set, and a deployment that never set the key loses only the settlement — the run stays