feat(session): hand off after final shell exit (#241)
- commit
- 0bf02ca4c2a59a2c72e8ef121d85c9918f39ddd8
- author
- bnema · · 2026-08-30 02:49 UTC
- committer
- GitHub · · 2026-08-30 02:49 UTC
- parent
- d73dd26
- signature
- GPG Unverified
B5690EEEBB952194The key is unknown
13 files · +323 -35
M README.md +1 -1
| @@ -38,7 +38,7 @@ vev host list list remote hosts | |||
| 38 | 38 | vev kill <name>|--all kill sessions | |
| 39 | 39 | ``` | |
| 40 | 40 | ||
| 41 | - | The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md). | |
| 41 | + | The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. Exiting the final shell removes that session and returns to the most recently used session; with no previous session, vev exits. See [durable-session recovery](docs/durable-session-recovery.md). | |
| 42 | 42 | ||
| 43 | 43 | > [!NOTE] | |
| 44 | 44 | > A release that bumps vev's protocol version resets named sessions saved by an older protocol. Session names are retained, but layouts, tabs, terminal history, recovery transcripts, and process-recovery state are discarded. Each local or remote daemon applies this reset to its own sessions when it first starts with the new protocol. | |
| @@ -38,7 +38,7 @@ vev host list list remote hosts | |||||
| 38 | vev kill <name>|--all kill sessions | 38 | vev kill <name>|--all kill sessions | ||
| 39 | ``` | 39 | ``` | ||
| 40 | 40 | ||||
| 41 | - | The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md). | 41 | + | The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. Exiting the final shell removes that session and returns to the most recently used session; with no previous session, vev exits. See [durable-session recovery](docs/durable-session-recovery.md). |
| 42 | 42 | ||||
| 43 | > [!NOTE] | 43 | > [!NOTE] | ||
| 44 | > A release that bumps vev's protocol version resets named sessions saved by an older protocol. Session names are retained, but layouts, tabs, terminal history, recovery transcripts, and process-recovery state are discarded. Each local or remote daemon applies this reset to its own sessions when it first starts with the new protocol. | 44 | > A release that bumps vev's protocol version resets named sessions saved by an older protocol. Session names are retained, but layouts, tabs, terminal history, recovery transcripts, and process-recovery state are discarded. Each local or remote daemon applies this reset to its own sessions when it first starts with the new protocol. | ||
M internal/usecase/client/client.go +62 -1
| @@ -468,6 +468,8 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 468 | 468 | routeNavigationPending := false | |
| 469 | 469 | routeNavigationResumeFallback := false | |
| 470 | 470 | var routeNavigationSelection *routeNavigationSelection | |
| 471 | + | var killedSelection *killedRouteSelection | |
| 472 | + | killedResumeFallback := false | |
| 471 | 473 | var routeNavigationFallback *attachRoute | |
| 472 | 474 | var routeNavigationAction *protocol.RouteNavigationAction | |
| 473 | 475 | creationPending := false | |
| @@ -633,6 +635,9 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 633 | 635 | if err != nil { | |
| 634 | 636 | err = handshakeContextError(ctx, timedOut, err) | |
| 635 | 637 | finishHandshake() | |
| 638 | + | if killedSelection != nil { | |
| 639 | + | return errors.Join(err, r.ledger.retireKilled(*killedSelection)) | |
| 640 | + | } | |
| 636 | 641 | if creationPending && creationFallback != nil && ctx.Err() == nil { | |
| 637 | 642 | r.creationFailure = &protocol.SessionCreationFailure{RequestID: creationRequestID, Code: routeFailureCode(err)} | |
| 638 | 643 | route := *creationFallback | |
| @@ -671,6 +676,9 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 671 | 676 | if transport == nil { | |
| 672 | 677 | finishHandshake() | |
| 673 | 678 | err := errors.New("vev: dialer returned nil transport") | |
| 679 | + | if killedSelection != nil { | |
| 680 | + | return errors.Join(err, r.ledger.retireKilled(*killedSelection)) | |
| 681 | + | } | |
| 674 | 682 | if creationPending && creationFallback != nil { | |
| 675 | 683 | r.creationFailure = &protocol.SessionCreationFailure{RequestID: creationRequestID, Code: protocol.RouteFailureUnavailable} | |
| 676 | 684 | route := *creationFallback | |
| @@ -721,6 +729,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 721 | 729 | request: attemptRequest, | |
| 722 | 730 | resumeToken: resumeToken, | |
| 723 | 731 | routeNavigationSelection: routeNavigationSelection, | |
| 732 | + | killedSelection: killedSelection, | |
| 724 | 733 | clientID: processClientID, | |
| 725 | 734 | milestones: &ms, | |
| 726 | 735 | themeState: themeState, | |
| @@ -772,6 +781,10 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 772 | 781 | creationFallback = nil | |
| 773 | 782 | creationRequestID = 0 | |
| 774 | 783 | } | |
| 784 | + | if result.err == nil && result.welcomed && killedSelection != nil { | |
| 785 | + | killedSelection = nil | |
| 786 | + | killedResumeFallback = false | |
| 787 | + | } | |
| 775 | 788 | if result.err == nil && result.welcomed && routeNavigationPending { | |
| 776 | 789 | routeNavigationPending = false | |
| 777 | 790 | routeNavigationSelection = nil | |
| @@ -982,6 +995,51 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 982 | 995 | } | |
| 983 | 996 | return nil | |
| 984 | 997 | } | |
| 998 | + | var detached *DetachedError | |
| 999 | + | if errors.As(result.err, &detached) && detached.Reason == protocol.ReasonSessionKilled { | |
| 1000 | + | if r.ledger == nil { | |
| 1001 | + | return errors.New("vev: route ledger unavailable after session kill") | |
| 1002 | + | } | |
| 1003 | + | selection, hasPrevious := r.ledger.killedSelection() | |
| 1004 | + | if !hasPrevious { | |
| 1005 | + | if err := r.ledger.retireKilled(selection); err != nil { | |
| 1006 | + | return err | |
| 1007 | + | } | |
| 1008 | + | return reconnect.clear() | |
| 1009 | + | } | |
| 1010 | + | killedSelection = &selection | |
| 1011 | + | dialer = selection.selected.dialer | |
| 1012 | + | attemptRequest = cloneAttachRequest(selection.selected.request) | |
| 1013 | + | attemptRequest.StartupOverlay = protocol.StartupOverlayNone | |
| 1014 | + | attemptRequest.NavigationCapabilities = 0 | |
| 1015 | + | resumeToken = selection.selected.resumeToken | |
| 1016 | + | killedResumeFallback = resumeToken != 0 | |
| 1017 | + | if resumeToken != 0 { | |
| 1018 | + | attemptRequest.Intent = protocol.IntentResume | |
| 1019 | + | } else { | |
| 1020 | + | attemptRequest.Intent = protocol.IntentAttach | |
| 1021 | + | } | |
| 1022 | + | homeRoute = nil | |
| 1023 | + | returnRoute = nil | |
| 1024 | + | homeNavigationPending = false | |
| 1025 | + | returnNavigationPending = false | |
| 1026 | + | routeNavigationPending = false | |
| 1027 | + | routeNavigationSelection = nil | |
| 1028 | + | routeNavigationFallback = nil | |
| 1029 | + | routeNavigationAction = nil | |
| 1030 | + | remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote) | |
| 1031 | + | backoff = defaultReconnectBackoff.initial | |
| 1032 | + | continue | |
| 1033 | + | } | |
| 1034 | + | if killedSelection != nil { | |
| 1035 | + | if killedResumeFallback && resumeNeedsExactAttach(result.err) { | |
| 1036 | + | attemptRequest.Intent = protocol.IntentAttach | |
| 1037 | + | resumeToken = 0 | |
| 1038 | + | killedResumeFallback = false | |
| 1039 | + | continue | |
| 1040 | + | } | |
| 1041 | + | return errors.Join(result.err, r.ledger.retireKilled(*killedSelection)) | |
| 1042 | + | } | |
| 985 | 1043 | if result.welcomed { | |
| 986 | 1044 | homeNavigationPending = false | |
| 987 | 1045 | } | |
| @@ -1205,6 +1263,7 @@ type attachAttempt struct { | |||
| 1205 | 1263 | request AttachRequest | |
| 1206 | 1264 | resumeToken uint64 | |
| 1207 | 1265 | routeNavigationSelection *routeNavigationSelection | |
| 1266 | + | killedSelection *killedRouteSelection | |
| 1208 | 1267 | clientID [16]byte | |
| 1209 | 1268 | milestones *milestones | |
| 1210 | 1269 | themeState *terminalThemeState | |
| @@ -1444,7 +1503,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||
| 1444 | 1503 | if !a.transientPicker { | |
| 1445 | 1504 | candidate := routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken) | |
| 1446 | 1505 | var commitErr error | |
| 1447 | - | if a.routeNavigationSelection != nil { | |
| 1506 | + | if a.killedSelection != nil { | |
| 1507 | + | commitErr = a.runner.ledger.commitKilledTransition(*a.killedSelection, candidate) | |
| 1508 | + | } else if a.routeNavigationSelection != nil { | |
| 1448 | 1509 | selection := *a.routeNavigationSelection | |
| 1449 | 1510 | commitErr = a.runner.ledger.commitTransition(selection.selected.identity, selection.snapshotGeneration, selection.active, candidate) | |
| 1450 | 1511 | } else { | |
| @@ -468,6 +468,8 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 468 | routeNavigationPending := false | 468 | routeNavigationPending := false | ||
| 469 | routeNavigationResumeFallback := false | 469 | routeNavigationResumeFallback := false | ||
| 470 | var routeNavigationSelection *routeNavigationSelection | 470 | var routeNavigationSelection *routeNavigationSelection | ||
| 471 | + | var killedSelection *killedRouteSelection | |||
| 472 | + | killedResumeFallback := false | |||
| 471 | var routeNavigationFallback *attachRoute | 473 | var routeNavigationFallback *attachRoute | ||
| 472 | var routeNavigationAction *protocol.RouteNavigationAction | 474 | var routeNavigationAction *protocol.RouteNavigationAction | ||
| 473 | creationPending := false | 475 | creationPending := false | ||
| @@ -633,6 +635,9 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 633 | if err != nil { | 635 | if err != nil { | ||
| 634 | err = handshakeContextError(ctx, timedOut, err) | 636 | err = handshakeContextError(ctx, timedOut, err) | ||
| 635 | finishHandshake() | 637 | finishHandshake() | ||
| 638 | + | if killedSelection != nil { | |||
| 639 | + | return errors.Join(err, r.ledger.retireKilled(*killedSelection)) | |||
| 640 | + | } | |||
| 636 | if creationPending && creationFallback != nil && ctx.Err() == nil { | 641 | if creationPending && creationFallback != nil && ctx.Err() == nil { | ||
| 637 | r.creationFailure = &protocol.SessionCreationFailure{RequestID: creationRequestID, Code: routeFailureCode(err)} | 642 | r.creationFailure = &protocol.SessionCreationFailure{RequestID: creationRequestID, Code: routeFailureCode(err)} | ||
| 638 | route := *creationFallback | 643 | route := *creationFallback | ||
| @@ -671,6 +676,9 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 671 | if transport == nil { | 676 | if transport == nil { | ||
| 672 | finishHandshake() | 677 | finishHandshake() | ||
| 673 | err := errors.New("vev: dialer returned nil transport") | 678 | err := errors.New("vev: dialer returned nil transport") | ||
| 679 | + | if killedSelection != nil { | |||
| 680 | + | return errors.Join(err, r.ledger.retireKilled(*killedSelection)) | |||
| 681 | + | } | |||
| 674 | if creationPending && creationFallback != nil { | 682 | if creationPending && creationFallback != nil { | ||
| 675 | r.creationFailure = &protocol.SessionCreationFailure{RequestID: creationRequestID, Code: protocol.RouteFailureUnavailable} | 683 | r.creationFailure = &protocol.SessionCreationFailure{RequestID: creationRequestID, Code: protocol.RouteFailureUnavailable} | ||
| 676 | route := *creationFallback | 684 | route := *creationFallback | ||
| @@ -721,6 +729,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 721 | request: attemptRequest, | 729 | request: attemptRequest, | ||
| 722 | resumeToken: resumeToken, | 730 | resumeToken: resumeToken, | ||
| 723 | routeNavigationSelection: routeNavigationSelection, | 731 | routeNavigationSelection: routeNavigationSelection, | ||
| 732 | + | killedSelection: killedSelection, | |||
| 724 | clientID: processClientID, | 733 | clientID: processClientID, | ||
| 725 | milestones: &ms, | 734 | milestones: &ms, | ||
| 726 | themeState: themeState, | 735 | themeState: themeState, | ||
| @@ -772,6 +781,10 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 772 | creationFallback = nil | 781 | creationFallback = nil | ||
| 773 | creationRequestID = 0 | 782 | creationRequestID = 0 | ||
| 774 | } | 783 | } | ||
| 784 | + | if result.err == nil && result.welcomed && killedSelection != nil { | |||
| 785 | + | killedSelection = nil | |||
| 786 | + | killedResumeFallback = false | |||
| 787 | + | } | |||
| 775 | if result.err == nil && result.welcomed && routeNavigationPending { | 788 | if result.err == nil && result.welcomed && routeNavigationPending { | ||
| 776 | routeNavigationPending = false | 789 | routeNavigationPending = false | ||
| 777 | routeNavigationSelection = nil | 790 | routeNavigationSelection = nil | ||
| @@ -982,6 +995,51 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 982 | } | 995 | } | ||
| 983 | return nil | 996 | return nil | ||
| 984 | } | 997 | } | ||
| 998 | + | var detached *DetachedError | |||
| 999 | + | if errors.As(result.err, &detached) && detached.Reason == protocol.ReasonSessionKilled { | |||
| 1000 | + | if r.ledger == nil { | |||
| 1001 | + | return errors.New("vev: route ledger unavailable after session kill") | |||
| 1002 | + | } | |||
| 1003 | + | selection, hasPrevious := r.ledger.killedSelection() | |||
| 1004 | + | if !hasPrevious { | |||
| 1005 | + | if err := r.ledger.retireKilled(selection); err != nil { | |||
| 1006 | + | return err | |||
| 1007 | + | } | |||
| 1008 | + | return reconnect.clear() | |||
| 1009 | + | } | |||
| 1010 | + | killedSelection = &selection | |||
| 1011 | + | dialer = selection.selected.dialer | |||
| 1012 | + | attemptRequest = cloneAttachRequest(selection.selected.request) | |||
| 1013 | + | attemptRequest.StartupOverlay = protocol.StartupOverlayNone | |||
| 1014 | + | attemptRequest.NavigationCapabilities = 0 | |||
| 1015 | + | resumeToken = selection.selected.resumeToken | |||
| 1016 | + | killedResumeFallback = resumeToken != 0 | |||
| 1017 | + | if resumeToken != 0 { | |||
| 1018 | + | attemptRequest.Intent = protocol.IntentResume | |||
| 1019 | + | } else { | |||
| 1020 | + | attemptRequest.Intent = protocol.IntentAttach | |||
| 1021 | + | } | |||
| 1022 | + | homeRoute = nil | |||
| 1023 | + | returnRoute = nil | |||
| 1024 | + | homeNavigationPending = false | |||
| 1025 | + | returnNavigationPending = false | |||
| 1026 | + | routeNavigationPending = false | |||
| 1027 | + | routeNavigationSelection = nil | |||
| 1028 | + | routeNavigationFallback = nil | |||
| 1029 | + | routeNavigationAction = nil | |||
| 1030 | + | remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote) | |||
| 1031 | + | backoff = defaultReconnectBackoff.initial | |||
| 1032 | + | continue | |||
| 1033 | + | } | |||
| 1034 | + | if killedSelection != nil { | |||
| 1035 | + | if killedResumeFallback && resumeNeedsExactAttach(result.err) { | |||
| 1036 | + | attemptRequest.Intent = protocol.IntentAttach | |||
| 1037 | + | resumeToken = 0 | |||
| 1038 | + | killedResumeFallback = false | |||
| 1039 | + | continue | |||
| 1040 | + | } | |||
| 1041 | + | return errors.Join(result.err, r.ledger.retireKilled(*killedSelection)) | |||
| 1042 | + | } | |||
| 985 | if result.welcomed { | 1043 | if result.welcomed { | ||
| 986 | homeNavigationPending = false | 1044 | homeNavigationPending = false | ||
| 987 | } | 1045 | } | ||
| @@ -1205,6 +1263,7 @@ type attachAttempt struct { | |||||
| 1205 | request AttachRequest | 1263 | request AttachRequest | ||
| 1206 | resumeToken uint64 | 1264 | resumeToken uint64 | ||
| 1207 | routeNavigationSelection *routeNavigationSelection | 1265 | routeNavigationSelection *routeNavigationSelection | ||
| 1266 | + | killedSelection *killedRouteSelection | |||
| 1208 | clientID [16]byte | 1267 | clientID [16]byte | ||
| 1209 | milestones *milestones | 1268 | milestones *milestones | ||
| 1210 | themeState *terminalThemeState | 1269 | themeState *terminalThemeState | ||
| @@ -1444,7 +1503,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||||
| 1444 | if !a.transientPicker { | 1503 | if !a.transientPicker { | ||
| 1445 | candidate := routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken) | 1504 | candidate := routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken) | ||
| 1446 | var commitErr error | 1505 | var commitErr error | ||
| 1447 | - | if a.routeNavigationSelection != nil { | 1506 | + | if a.killedSelection != nil { |
| 1507 | + | commitErr = a.runner.ledger.commitKilledTransition(*a.killedSelection, candidate) | |||
| 1508 | + | } else if a.routeNavigationSelection != nil { | |||
| 1448 | selection := *a.routeNavigationSelection | 1509 | selection := *a.routeNavigationSelection | ||
| 1449 | commitErr = a.runner.ledger.commitTransition(selection.selected.identity, selection.snapshotGeneration, selection.active, candidate) | 1510 | commitErr = a.runner.ledger.commitTransition(selection.selected.identity, selection.snapshotGeneration, selection.active, candidate) | ||
| 1450 | } else { | 1511 | } else { | ||
M internal/usecase/client/client_test.go +43 -4
| @@ -515,6 +515,47 @@ func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) { | |||
| 515 | 515 | require.Equal(t, int32(1), dialer.calls.Load(), "same-peer switching must not dial a replacement transport") | |
| 516 | 516 | } | |
| 517 | 517 | ||
| 518 | + | func TestKilledSessionReturnsToPreviousLocalRoute(t *testing.T) { | |
| 519 | + | term := newRunTerminal() | |
| 520 | + | defer term.in.unblock() | |
| 521 | + | ||
| 522 | + | firstTarget := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "first"} | |
| 523 | + | secondTarget := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{2}, SessionName: "second"} | |
| 524 | + | welcome := func(target protocol.ExactSessionTarget) wire.Frame { | |
| 525 | + | return frameOf(wire.MsgWelcome, wire.MarshalWelcome(protocol.Welcome{ | |
| 526 | + | SessionID: target.SessionName, SessionName: target.SessionName, ResumeToken: 1, | |
| 527 | + | Capabilities: protocol.CapabilityResume, | |
| 528 | + | CommittedIdentity: &protocol.CommittedRouteIdentity{Target: target}, | |
| 529 | + | })) | |
| 530 | + | } | |
| 531 | + | active := &recordingTransport{recvs: []recvItem{ | |
| 532 | + | {f: welcome(firstTarget)}, | |
| 533 | + | {f: frameOf(wire.MsgAttachTarget, wire.MarshalAttachTarget(protocol.AttachTarget{ | |
| 534 | + | Session: "second", Intent: protocol.IntentAttach, ExactTarget: &secondTarget, | |
| 535 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |
| 536 | + | }))}, | |
| 537 | + | {f: frameOf(wire.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(protocol.CommittedRouteIdentity{Target: secondTarget}))}, | |
| 538 | + | {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonSessionKilled}))}, | |
| 539 | + | }} | |
| 540 | + | previous := &recordingTransport{recvs: []recvItem{ | |
| 541 | + | {f: welcome(firstTarget)}, | |
| 542 | + | {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach}))}, | |
| 543 | + | }} | |
| 544 | + | dialer := &sequenceDialer{trs: []wire.Transport{active, previous}} | |
| 545 | + | ||
| 546 | + | err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{ | |
| 547 | + | Intent: protocol.IntentAttach, SessionName: "first", Origin: protocol.RouteOriginLocal, | |
| 548 | + | OriginKey: "local", EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned, | |
| 549 | + | }) | |
| 550 | + | ||
| 551 | + | require.NoError(t, err) | |
| 552 | + | require.Equal(t, int32(2), dialer.calls.Load()) | |
| 553 | + | hello := helloFromSend(t, previous) | |
| 554 | + | require.Equal(t, protocol.IntentResume, hello.Intent) | |
| 555 | + | require.Equal(t, firstTarget, *hello.ExactTarget) | |
| 556 | + | require.Equal(t, int32(1), term.restoreCount.Load()) | |
| 557 | + | } | |
| 558 | + | ||
| 518 | 559 | func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) { | |
| 519 | 560 | term := newRunTerminal() | |
| 520 | 561 | defer term.in.unblock() | |
| @@ -2497,16 +2538,14 @@ func TestRunReturnsWhenRemoteHostLearnerStalls(t *testing.T) { | |||
| 2497 | 2538 | } | |
| 2498 | 2539 | } | |
| 2499 | 2540 | ||
| 2500 | - | func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) { | |
| 2541 | + | func TestRunExitsCleanlyWhenKilledSessionHasNoPriorRoute(t *testing.T) { | |
| 2501 | 2542 | term := newRunTerminal() | |
| 2502 | 2543 | defer term.in.unblock() | |
| 2503 | 2544 | tr := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonSessionKilled}))}}} | |
| 2504 | 2545 | d := &sequenceDialer{trs: []wire.Transport{tr}} | |
| 2505 | 2546 | ||
| 2506 | 2547 | err := runTestClient(context.Background(), testDependencies(d, term, realClock{}, nil, nil), client.AttachRequest{Intent: protocol.IntentAttach, SessionName: "main", Remote: false}) | |
| 2507 | - | require.Error(t, err) | |
| 2508 | - | var de *client.DetachedError | |
| 2509 | - | require.True(t, errors.As(err, &de)) | |
| 2548 | + | require.NoError(t, err) | |
| 2510 | 2549 | require.Equal(t, int32(1), d.calls.Load()) | |
| 2511 | 2550 | require.Equal(t, int32(1), term.restoreCount.Load()) | |
| 2512 | 2551 | } | |
| @@ -515,6 +515,47 @@ func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) { | |||||
| 515 | require.Equal(t, int32(1), dialer.calls.Load(), "same-peer switching must not dial a replacement transport") | 515 | require.Equal(t, int32(1), dialer.calls.Load(), "same-peer switching must not dial a replacement transport") | ||
| 516 | } | 516 | } | ||
| 517 | 517 | ||||
| 518 | + | func TestKilledSessionReturnsToPreviousLocalRoute(t *testing.T) { | |||
| 519 | + | term := newRunTerminal() | |||
| 520 | + | defer term.in.unblock() | |||
| 521 | + | ||||
| 522 | + | firstTarget := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "first"} | |||
| 523 | + | secondTarget := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{2}, SessionName: "second"} | |||
| 524 | + | welcome := func(target protocol.ExactSessionTarget) wire.Frame { | |||
| 525 | + | return frameOf(wire.MsgWelcome, wire.MarshalWelcome(protocol.Welcome{ | |||
| 526 | + | SessionID: target.SessionName, SessionName: target.SessionName, ResumeToken: 1, | |||
| 527 | + | Capabilities: protocol.CapabilityResume, | |||
| 528 | + | CommittedIdentity: &protocol.CommittedRouteIdentity{Target: target}, | |||
| 529 | + | })) | |||
| 530 | + | } | |||
| 531 | + | active := &recordingTransport{recvs: []recvItem{ | |||
| 532 | + | {f: welcome(firstTarget)}, | |||
| 533 | + | {f: frameOf(wire.MsgAttachTarget, wire.MarshalAttachTarget(protocol.AttachTarget{ | |||
| 534 | + | Session: "second", Intent: protocol.IntentAttach, ExactTarget: &secondTarget, | |||
| 535 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |||
| 536 | + | }))}, | |||
| 537 | + | {f: frameOf(wire.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(protocol.CommittedRouteIdentity{Target: secondTarget}))}, | |||
| 538 | + | {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonSessionKilled}))}, | |||
| 539 | + | }} | |||
| 540 | + | previous := &recordingTransport{recvs: []recvItem{ | |||
| 541 | + | {f: welcome(firstTarget)}, | |||
| 542 | + | {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach}))}, | |||
| 543 | + | }} | |||
| 544 | + | dialer := &sequenceDialer{trs: []wire.Transport{active, previous}} | |||
| 545 | + | ||||
| 546 | + | err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{ | |||
| 547 | + | Intent: protocol.IntentAttach, SessionName: "first", Origin: protocol.RouteOriginLocal, | |||
| 548 | + | OriginKey: "local", EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned, | |||
| 549 | + | }) | |||
| 550 | + | ||||
| 551 | + | require.NoError(t, err) | |||
| 552 | + | require.Equal(t, int32(2), dialer.calls.Load()) | |||
| 553 | + | hello := helloFromSend(t, previous) | |||
| 554 | + | require.Equal(t, protocol.IntentResume, hello.Intent) | |||
| 555 | + | require.Equal(t, firstTarget, *hello.ExactTarget) | |||
| 556 | + | require.Equal(t, int32(1), term.restoreCount.Load()) | |||
| 557 | + | } | |||
| 558 | + | ||||
| 518 | func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) { | 559 | func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) { | ||
| 519 | term := newRunTerminal() | 560 | term := newRunTerminal() | ||
| 520 | defer term.in.unblock() | 561 | defer term.in.unblock() | ||
| @@ -2497,16 +2538,14 @@ func TestRunReturnsWhenRemoteHostLearnerStalls(t *testing.T) { | |||||
| 2497 | } | 2538 | } | ||
| 2498 | } | 2539 | } | ||
| 2499 | 2540 | ||||
| 2500 | - | func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) { | 2541 | + | func TestRunExitsCleanlyWhenKilledSessionHasNoPriorRoute(t *testing.T) { |
| 2501 | term := newRunTerminal() | 2542 | term := newRunTerminal() | ||
| 2502 | defer term.in.unblock() | 2543 | defer term.in.unblock() | ||
| 2503 | tr := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonSessionKilled}))}}} | 2544 | tr := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonSessionKilled}))}}} | ||
| 2504 | d := &sequenceDialer{trs: []wire.Transport{tr}} | 2545 | d := &sequenceDialer{trs: []wire.Transport{tr}} | ||
| 2505 | 2546 | ||||
| 2506 | err := runTestClient(context.Background(), testDependencies(d, term, realClock{}, nil, nil), client.AttachRequest{Intent: protocol.IntentAttach, SessionName: "main", Remote: false}) | 2547 | err := runTestClient(context.Background(), testDependencies(d, term, realClock{}, nil, nil), client.AttachRequest{Intent: protocol.IntentAttach, SessionName: "main", Remote: false}) | ||
| 2507 | - | require.Error(t, err) | 2548 | + | require.NoError(t, err) |
| 2508 | - | var de *client.DetachedError | |||
| 2509 | - | require.True(t, errors.As(err, &de)) | |||
| 2510 | require.Equal(t, int32(1), d.calls.Load()) | 2549 | require.Equal(t, int32(1), d.calls.Load()) | ||
| 2511 | require.Equal(t, int32(1), term.restoreCount.Load()) | 2550 | require.Equal(t, int32(1), term.restoreCount.Load()) | ||
| 2512 | } | 2551 | } | ||
M internal/usecase/client/reconnect_toast_test.go +2 -5
| @@ -1678,11 +1678,8 @@ func TestRemoteReconnectToastLifecycle(t *testing.T) { | |||
| 1678 | 1678 | dialer.EXPECT().Dial(mock.Anything).Return(tr2, nil).Once() | |
| 1679 | 1679 | return []*reconnectToastRecordingTransport{tr1, tr2} | |
| 1680 | 1680 | }, | |
| 1681 | - | sleep: func(context.CancelFunc) bool { return true }, | |
| 1682 | - | wantErr: func(t *testing.T, err error) { | |
| 1683 | - | var detached *DetachedError | |
| 1684 | - | require.True(t, errors.As(err, &detached)) | |
| 1685 | - | }, | |
| 1681 | + | sleep: func(context.CancelFunc) bool { return true }, | |
| 1682 | + | wantErr: func(t *testing.T, err error) { require.NoError(t, err) }, | |
| 1686 | 1683 | }, | |
| 1687 | 1684 | } | |
| 1688 | 1685 | ||
| @@ -1678,11 +1678,8 @@ func TestRemoteReconnectToastLifecycle(t *testing.T) { | |||||
| 1678 | dialer.EXPECT().Dial(mock.Anything).Return(tr2, nil).Once() | 1678 | dialer.EXPECT().Dial(mock.Anything).Return(tr2, nil).Once() | ||
| 1679 | return []*reconnectToastRecordingTransport{tr1, tr2} | 1679 | return []*reconnectToastRecordingTransport{tr1, tr2} | ||
| 1680 | }, | 1680 | }, | ||
| 1681 | - | sleep: func(context.CancelFunc) bool { return true }, | 1681 | + | sleep: func(context.CancelFunc) bool { return true }, |
| 1682 | - | wantErr: func(t *testing.T, err error) { | 1682 | + | wantErr: func(t *testing.T, err error) { require.NoError(t, err) }, |
| 1683 | - | var detached *DetachedError | |||
| 1684 | - | require.True(t, errors.As(err, &detached)) | |||
| 1685 | - | }, | |||
| 1686 | }, | 1683 | }, | ||
| 1687 | } | 1684 | } | ||
| 1688 | 1685 | ||||
M internal/usecase/client/routes.go +75 -0
| @@ -646,6 +646,81 @@ type routeNavigationSelection struct { | |||
| 646 | 646 | noOp bool | |
| 647 | 647 | } | |
| 648 | 648 | ||
| 649 | + | type killedRouteSelection struct { | |
| 650 | + | selected routeRecord | |
| 651 | + | sourceAliases []routeIdentity | |
| 652 | + | snapshotGeneration routeGeneration | |
| 653 | + | active routeIdentity | |
| 654 | + | } | |
| 655 | + | ||
| 656 | + | // killedSelection captures the newest route that does not identify the active | |
| 657 | + | // endpoint authority and exact lifecycle. Route origin is intentionally not | |
| 658 | + | // part of alias equivalence: direct and discovery routes can retain the same | |
| 659 | + | // canonical endpoint authority. | |
| 660 | + | func (l *routeLedger) killedSelection() (killedRouteSelection, bool) { | |
| 661 | + | l.mu.RLock() | |
| 662 | + | defer l.mu.RUnlock() | |
| 663 | + | selection := killedRouteSelection{snapshotGeneration: l.generation, active: l.active} | |
| 664 | + | activeIndex := l.indexByIdentityLocked(l.active) | |
| 665 | + | if activeIndex < 0 { | |
| 666 | + | return selection, false | |
| 667 | + | } | |
| 668 | + | source := l.entries[activeIndex] | |
| 669 | + | for _, entry := range l.entries { | |
| 670 | + | if entry.originKey == source.originKey && entry.target.LifecycleID == source.target.LifecycleID { | |
| 671 | + | selection.sourceAliases = append(selection.sourceAliases, entry.identity) | |
| 672 | + | continue | |
| 673 | + | } | |
| 674 | + | if selection.selected.identity.empty() { | |
| 675 | + | selection.selected = cloneRouteRecord(entry) | |
| 676 | + | } | |
| 677 | + | } | |
| 678 | + | return selection, !selection.selected.identity.empty() | |
| 679 | + | } | |
| 680 | + | ||
| 681 | + | func (l *routeLedger) retireKilled(selection killedRouteSelection) error { | |
| 682 | + | l.mu.Lock() | |
| 683 | + | defer l.mu.Unlock() | |
| 684 | + | if l.generation != selection.snapshotGeneration || l.active != selection.active { | |
| 685 | + | return errRouteStaleSelection | |
| 686 | + | } | |
| 687 | + | l.removeKilledAliasesLocked(selection.sourceAliases) | |
| 688 | + | l.generation++ | |
| 689 | + | return nil | |
| 690 | + | } | |
| 691 | + | ||
| 692 | + | func (l *routeLedger) commitKilledTransition(selection killedRouteSelection, candidate routeCandidate) error { | |
| 693 | + | candidate, err := prepareRouteCandidate(candidate) | |
| 694 | + | if err != nil { | |
| 695 | + | return err | |
| 696 | + | } | |
| 697 | + | l.mu.Lock() | |
| 698 | + | defer l.mu.Unlock() | |
| 699 | + | if l.generation != selection.snapshotGeneration || l.active != selection.active || | |
| 700 | + | l.indexByIdentityLocked(selection.selected.identity) < 0 { | |
| 701 | + | return errRouteStaleSelection | |
| 702 | + | } | |
| 703 | + | l.removeKilledAliasesLocked(selection.sourceAliases) | |
| 704 | + | if _, err = l.commitLocked(candidate); err != nil { | |
| 705 | + | return err | |
| 706 | + | } | |
| 707 | + | if l.previous.empty() && len(l.entries) > 1 { | |
| 708 | + | l.previous = l.entries[1].identity | |
| 709 | + | } | |
| 710 | + | return nil | |
| 711 | + | } | |
| 712 | + | ||
| 713 | + | func (l *routeLedger) removeKilledAliasesLocked(aliases []routeIdentity) { | |
| 714 | + | for i := len(l.entries) - 1; i >= 0; i-- { | |
| 715 | + | for _, alias := range aliases { | |
| 716 | + | if l.entries[i].identity == alias { | |
| 717 | + | l.removeAtLocked(i) | |
| 718 | + | break | |
| 719 | + | } | |
| 720 | + | } | |
| 721 | + | } | |
| 722 | + | } | |
| 723 | + | ||
| 649 | 724 | // navigationSelection validates an action against the latest complete | |
| 650 | 725 | // snapshot and captures the selected route, live restoration route, snapshot | |
| 651 | 726 | // generation, and active identity under one read lock. The commit phase uses | |
| @@ -646,6 +646,81 @@ type routeNavigationSelection struct { | |||||
| 646 | noOp bool | 646 | noOp bool | ||
| 647 | } | 647 | } | ||
| 648 | 648 | ||||
| 649 | + | type killedRouteSelection struct { | |||
| 650 | + | selected routeRecord | |||
| 651 | + | sourceAliases []routeIdentity | |||
| 652 | + | snapshotGeneration routeGeneration | |||
| 653 | + | active routeIdentity | |||
| 654 | + | } | |||
| 655 | + | ||||
| 656 | + | // killedSelection captures the newest route that does not identify the active | |||
| 657 | + | // endpoint authority and exact lifecycle. Route origin is intentionally not | |||
| 658 | + | // part of alias equivalence: direct and discovery routes can retain the same | |||
| 659 | + | // canonical endpoint authority. | |||
| 660 | + | func (l *routeLedger) killedSelection() (killedRouteSelection, bool) { | |||
| 661 | + | l.mu.RLock() | |||
| 662 | + | defer l.mu.RUnlock() | |||
| 663 | + | selection := killedRouteSelection{snapshotGeneration: l.generation, active: l.active} | |||
| 664 | + | activeIndex := l.indexByIdentityLocked(l.active) | |||
| 665 | + | if activeIndex < 0 { | |||
| 666 | + | return selection, false | |||
| 667 | + | } | |||
| 668 | + | source := l.entries[activeIndex] | |||
| 669 | + | for _, entry := range l.entries { | |||
| 670 | + | if entry.originKey == source.originKey && entry.target.LifecycleID == source.target.LifecycleID { | |||
| 671 | + | selection.sourceAliases = append(selection.sourceAliases, entry.identity) | |||
| 672 | + | continue | |||
| 673 | + | } | |||
| 674 | + | if selection.selected.identity.empty() { | |||
| 675 | + | selection.selected = cloneRouteRecord(entry) | |||
| 676 | + | } | |||
| 677 | + | } | |||
| 678 | + | return selection, !selection.selected.identity.empty() | |||
| 679 | + | } | |||
| 680 | + | ||||
| 681 | + | func (l *routeLedger) retireKilled(selection killedRouteSelection) error { | |||
| 682 | + | l.mu.Lock() | |||
| 683 | + | defer l.mu.Unlock() | |||
| 684 | + | if l.generation != selection.snapshotGeneration || l.active != selection.active { | |||
| 685 | + | return errRouteStaleSelection | |||
| 686 | + | } | |||
| 687 | + | l.removeKilledAliasesLocked(selection.sourceAliases) | |||
| 688 | + | l.generation++ | |||
| 689 | + | return nil | |||
| 690 | + | } | |||
| 691 | + | ||||
| 692 | + | func (l *routeLedger) commitKilledTransition(selection killedRouteSelection, candidate routeCandidate) error { | |||
| 693 | + | candidate, err := prepareRouteCandidate(candidate) | |||
| 694 | + | if err != nil { | |||
| 695 | + | return err | |||
| 696 | + | } | |||
| 697 | + | l.mu.Lock() | |||
| 698 | + | defer l.mu.Unlock() | |||
| 699 | + | if l.generation != selection.snapshotGeneration || l.active != selection.active || | |||
| 700 | + | l.indexByIdentityLocked(selection.selected.identity) < 0 { | |||
| 701 | + | return errRouteStaleSelection | |||
| 702 | + | } | |||
| 703 | + | l.removeKilledAliasesLocked(selection.sourceAliases) | |||
| 704 | + | if _, err = l.commitLocked(candidate); err != nil { | |||
| 705 | + | return err | |||
| 706 | + | } | |||
| 707 | + | if l.previous.empty() && len(l.entries) > 1 { | |||
| 708 | + | l.previous = l.entries[1].identity | |||
| 709 | + | } | |||
| 710 | + | return nil | |||
| 711 | + | } | |||
| 712 | + | ||||
| 713 | + | func (l *routeLedger) removeKilledAliasesLocked(aliases []routeIdentity) { | |||
| 714 | + | for i := len(l.entries) - 1; i >= 0; i-- { | |||
| 715 | + | for _, alias := range aliases { | |||
| 716 | + | if l.entries[i].identity == alias { | |||
| 717 | + | l.removeAtLocked(i) | |||
| 718 | + | break | |||
| 719 | + | } | |||
| 720 | + | } | |||
| 721 | + | } | |||
| 722 | + | } | |||
| 723 | + | ||||
| 649 | // navigationSelection validates an action against the latest complete | 724 | // navigationSelection validates an action against the latest complete | ||
| 650 | // snapshot and captures the selected route, live restoration route, snapshot | 725 | // snapshot and captures the selected route, live restoration route, snapshot | ||
| 651 | // generation, and active identity under one read lock. The commit phase uses | 726 | // generation, and active identity under one read lock. The commit phase uses | ||
M internal/usecase/client/routes_test.go +58 -0
| @@ -33,6 +33,64 @@ func routeTestCandidate(index byte, origin protocol.RouteOrigin) routeCandidate | |||
| 33 | 33 | } | |
| 34 | 34 | } | |
| 35 | 35 | ||
| 36 | + | func TestRouteLedgerKilledSelectionRetiresSourceAliasesAndCommitsPrevious(t *testing.T) { | |
| 37 | + | ledger := newRouteLedger() | |
| 38 | + | older := routeTestCandidate(0, protocol.RouteOriginLocal) | |
| 39 | + | older.originKey = "local" | |
| 40 | + | _, err := ledger.commit(older) | |
| 41 | + | require.NoError(t, err) | |
| 42 | + | prior := routeTestCandidate(1, protocol.RouteOriginLocal) | |
| 43 | + | prior.originKey = "local" | |
| 44 | + | prior.request.PreferredTabID = "tab-prior" | |
| 45 | + | _, err = ledger.commit(prior) | |
| 46 | + | require.NoError(t, err) | |
| 47 | + | ||
| 48 | + | alias := routeTestCandidate(2, protocol.RouteOriginRemote) | |
| 49 | + | alias.originKey = "ssh://host" | |
| 50 | + | _, err = ledger.commit(alias) | |
| 51 | + | require.NoError(t, err) | |
| 52 | + | active := alias | |
| 53 | + | active.origin = protocol.RouteOriginDiscovery | |
| 54 | + | active.target.SessionName = "renamed-session" | |
| 55 | + | active.presentation.name = active.target.SessionName | |
| 56 | + | active.request.SessionName = active.target.SessionName | |
| 57 | + | _, err = ledger.commit(active) | |
| 58 | + | require.NoError(t, err) | |
| 59 | + | ||
| 60 | + | selection, ok := ledger.killedSelection() | |
| 61 | + | require.True(t, ok) | |
| 62 | + | require.Equal(t, prior.target, selection.selected.target) | |
| 63 | + | ||
| 64 | + | candidate := routeCandidateForAttach(selection.selected.request, | |
| 65 | + | protocol.CommittedRouteIdentity{Target: prior.target}, selection.selected.dialer, 0) | |
| 66 | + | require.NoError(t, ledger.commitKilledTransition(selection, candidate)) | |
| 67 | + | ||
| 68 | + | snapshot := ledger.snapshot() | |
| 69 | + | require.Equal(t, prior.target, snapshot.ActiveEntry.Target) | |
| 70 | + | require.Len(t, snapshot.Entries, 1) | |
| 71 | + | require.Equal(t, older.target, snapshot.Entries[0].Target) | |
| 72 | + | require.Equal(t, snapshot.Entries[0].Key, snapshot.Previous.Key) | |
| 73 | + | require.Equal(t, domain.TabStableID("tab-prior"), ledger.entries[0].request.PreferredTabID) | |
| 74 | + | for _, entry := range ledger.entries { | |
| 75 | + | require.NotEqual(t, alias.target, entry.target) | |
| 76 | + | } | |
| 77 | + | } | |
| 78 | + | ||
| 79 | + | func TestRouteLedgerKilledSelectionWithoutPreviousRetiresActive(t *testing.T) { | |
| 80 | + | ledger := newRouteLedger() | |
| 81 | + | active := routeTestCandidate(1, protocol.RouteOriginRemote) | |
| 82 | + | active.originKey = "ssh://host" | |
| 83 | + | _, err := ledger.commit(active) | |
| 84 | + | require.NoError(t, err) | |
| 85 | + | ||
| 86 | + | selection, ok := ledger.killedSelection() | |
| 87 | + | require.False(t, ok) | |
| 88 | + | require.NoError(t, ledger.retireKilled(selection)) | |
| 89 | + | require.Empty(t, ledger.entries) | |
| 90 | + | require.Empty(t, ledger.activeRef()) | |
| 91 | + | require.Empty(t, ledger.homeRef()) | |
| 92 | + | } | |
| 93 | + | ||
| 36 | 94 | func TestRouteCandidateRetainsHostLabelWithoutDiscoveryTarget(t *testing.T) { | |
| 37 | 95 | target := routeTestTarget(1) | |
| 38 | 96 | candidate := routeCandidateForAttach(AttachRequest{ | |
| @@ -33,6 +33,64 @@ func routeTestCandidate(index byte, origin protocol.RouteOrigin) routeCandidate | |||||
| 33 | } | 33 | } | ||
| 34 | } | 34 | } | ||
| 35 | 35 | ||||
| 36 | + | func TestRouteLedgerKilledSelectionRetiresSourceAliasesAndCommitsPrevious(t *testing.T) { | |||
| 37 | + | ledger := newRouteLedger() | |||
| 38 | + | older := routeTestCandidate(0, protocol.RouteOriginLocal) | |||
| 39 | + | older.originKey = "local" | |||
| 40 | + | _, err := ledger.commit(older) | |||
| 41 | + | require.NoError(t, err) | |||
| 42 | + | prior := routeTestCandidate(1, protocol.RouteOriginLocal) | |||
| 43 | + | prior.originKey = "local" | |||
| 44 | + | prior.request.PreferredTabID = "tab-prior" | |||
| 45 | + | _, err = ledger.commit(prior) | |||
| 46 | + | require.NoError(t, err) | |||
| 47 | + | ||||
| 48 | + | alias := routeTestCandidate(2, protocol.RouteOriginRemote) | |||
| 49 | + | alias.originKey = "ssh://host" | |||
| 50 | + | _, err = ledger.commit(alias) | |||
| 51 | + | require.NoError(t, err) | |||
| 52 | + | active := alias | |||
| 53 | + | active.origin = protocol.RouteOriginDiscovery | |||
| 54 | + | active.target.SessionName = "renamed-session" | |||
| 55 | + | active.presentation.name = active.target.SessionName | |||
| 56 | + | active.request.SessionName = active.target.SessionName | |||
| 57 | + | _, err = ledger.commit(active) | |||
| 58 | + | require.NoError(t, err) | |||
| 59 | + | ||||
| 60 | + | selection, ok := ledger.killedSelection() | |||
| 61 | + | require.True(t, ok) | |||
| 62 | + | require.Equal(t, prior.target, selection.selected.target) | |||
| 63 | + | ||||
| 64 | + | candidate := routeCandidateForAttach(selection.selected.request, | |||
| 65 | + | protocol.CommittedRouteIdentity{Target: prior.target}, selection.selected.dialer, 0) | |||
| 66 | + | require.NoError(t, ledger.commitKilledTransition(selection, candidate)) | |||
| 67 | + | ||||
| 68 | + | snapshot := ledger.snapshot() | |||
| 69 | + | require.Equal(t, prior.target, snapshot.ActiveEntry.Target) | |||
| 70 | + | require.Len(t, snapshot.Entries, 1) | |||
| 71 | + | require.Equal(t, older.target, snapshot.Entries[0].Target) | |||
| 72 | + | require.Equal(t, snapshot.Entries[0].Key, snapshot.Previous.Key) | |||
| 73 | + | require.Equal(t, domain.TabStableID("tab-prior"), ledger.entries[0].request.PreferredTabID) | |||
| 74 | + | for _, entry := range ledger.entries { | |||
| 75 | + | require.NotEqual(t, alias.target, entry.target) | |||
| 76 | + | } | |||
| 77 | + | } | |||
| 78 | + | ||||
| 79 | + | func TestRouteLedgerKilledSelectionWithoutPreviousRetiresActive(t *testing.T) { | |||
| 80 | + | ledger := newRouteLedger() | |||
| 81 | + | active := routeTestCandidate(1, protocol.RouteOriginRemote) | |||
| 82 | + | active.originKey = "ssh://host" | |||
| 83 | + | _, err := ledger.commit(active) | |||
| 84 | + | require.NoError(t, err) | |||
| 85 | + | ||||
| 86 | + | selection, ok := ledger.killedSelection() | |||
| 87 | + | require.False(t, ok) | |||
| 88 | + | require.NoError(t, ledger.retireKilled(selection)) | |||
| 89 | + | require.Empty(t, ledger.entries) | |||
| 90 | + | require.Empty(t, ledger.activeRef()) | |||
| 91 | + | require.Empty(t, ledger.homeRef()) | |||
| 92 | + | } | |||
| 93 | + | ||||
| 36 | func TestRouteCandidateRetainsHostLabelWithoutDiscoveryTarget(t *testing.T) { | 94 | func TestRouteCandidateRetainsHostLabelWithoutDiscoveryTarget(t *testing.T) { | ||
| 37 | target := routeTestTarget(1) | 95 | target := routeTestTarget(1) | ||
| 38 | candidate := routeCandidateForAttach(AttachRequest{ | 96 | candidate := routeCandidateForAttach(AttachRequest{ | ||
M internal/usecase/daemon/control_test.go +1 -0
| @@ -551,6 +551,7 @@ func TestCloseCommandsReportMutationOutcome(t *testing.T) { | |||
| 551 | 551 | t.Run("failed final pane close fails and retains session", func(t *testing.T) { | |
| 552 | 552 | d := newTestDaemon(t, &controlPTYFactory{}, stubClock{}) | |
| 553 | 553 | d.snapsEnabled = true | |
| 554 | + | d.persistEnabled = true | |
| 554 | 555 | d.snapshotRepository = portsmocks.NewMockSnapshotRepository(t) | |
| 555 | 556 | sess := addControlSession(d, "work", "t_work", "p_work") | |
| 556 | 557 | sess.mu.Lock() | |
| @@ -551,6 +551,7 @@ func TestCloseCommandsReportMutationOutcome(t *testing.T) { | |||||
| 551 | t.Run("failed final pane close fails and retains session", func(t *testing.T) { | 551 | t.Run("failed final pane close fails and retains session", func(t *testing.T) { | ||
| 552 | d := newTestDaemon(t, &controlPTYFactory{}, stubClock{}) | 552 | d := newTestDaemon(t, &controlPTYFactory{}, stubClock{}) | ||
| 553 | d.snapsEnabled = true | 553 | d.snapsEnabled = true | ||
| 554 | + | d.persistEnabled = true | |||
| 554 | d.snapshotRepository = portsmocks.NewMockSnapshotRepository(t) | 555 | d.snapshotRepository = portsmocks.NewMockSnapshotRepository(t) | ||
| 555 | sess := addControlSession(d, "work", "t_work", "p_work") | 556 | sess := addControlSession(d, "work", "t_work", "p_work") | ||
| 556 | sess.mu.Lock() | 557 | sess.mu.Lock() | ||
M internal/usecase/daemon/move_postcommit.go +1 -1
| @@ -160,7 +160,7 @@ func (p movePostcommitPlan) execute(d *Daemon) { | |||
| 160 | 160 | d.finishParkedAttachmentRetirements(p.retiredParked) | |
| 161 | 161 | if p.sourceEmpty && p.sourceMetadataValid && d.persistEnabled { | |
| 162 | 162 | if err := d.beginSnapshotPurge(p.sourceName, p.source.incarnation); err == nil { | |
| 163 | - | if err := d.finishSnapshotPurge(d.serveCtx, p.sourceName, p.source.incarnation); err != nil { | |
| 163 | + | if err := d.finishSnapshotPurge(d.serveCtx, p.sourceName, p.source.incarnation, p.source.createdAt); err != nil { | |
| 164 | 164 | d.log.Warn("moving final pane source purge failed", "err", err, "session", p.sourceName) | |
| 165 | 165 | } | |
| 166 | 166 | } | |
| @@ -160,7 +160,7 @@ func (p movePostcommitPlan) execute(d *Daemon) { | |||||
| 160 | d.finishParkedAttachmentRetirements(p.retiredParked) | 160 | d.finishParkedAttachmentRetirements(p.retiredParked) | ||
| 161 | if p.sourceEmpty && p.sourceMetadataValid && d.persistEnabled { | 161 | if p.sourceEmpty && p.sourceMetadataValid && d.persistEnabled { | ||
| 162 | if err := d.beginSnapshotPurge(p.sourceName, p.source.incarnation); err == nil { | 162 | if err := d.beginSnapshotPurge(p.sourceName, p.source.incarnation); err == nil { | ||
| 163 | - | if err := d.finishSnapshotPurge(d.serveCtx, p.sourceName, p.source.incarnation); err != nil { | 163 | + | if err := d.finishSnapshotPurge(d.serveCtx, p.sourceName, p.source.incarnation, p.source.createdAt); err != nil { |
| 164 | d.log.Warn("moving final pane source purge failed", "err", err, "session", p.sourceName) | 164 | d.log.Warn("moving final pane source purge failed", "err", err, "session", p.sourceName) | ||
| 165 | } | 165 | } | ||
| 166 | } | 166 | } | ||
M internal/usecase/daemon/session.go +16 -15
| @@ -1165,9 +1165,9 @@ func (d *Daemon) closeTabLockedWithEffect(sess *session, tb *tab, repaint bool, | |||
| 1165 | 1165 | var err error | |
| 1166 | 1166 | if effect != nil { | |
| 1167 | 1167 | effect.bindActionEnd(d, "close-tab") | |
| 1168 | - | err = d.killSessionForAttachment(sess, protocol.ReasonSessionKilled, false, effect, "close-tab") | |
| 1168 | + | err = d.killSessionForAttachment(sess, protocol.ReasonSessionKilled, true, effect, "close-tab") | |
| 1169 | 1169 | } else { | |
| 1170 | - | err = d.killSession(sess, protocol.ReasonSessionKilled, false) | |
| 1170 | + | err = d.killSession(sess, protocol.ReasonSessionKilled, true) | |
| 1171 | 1171 | } | |
| 1172 | 1172 | if err != nil { | |
| 1173 | 1173 | d.log.Warn("closing last tab failed", "session", name, "err", err) | |
| @@ -1240,7 +1240,7 @@ func (d *Daemon) beginSnapshotPurge(_ string, _ domain.IncarnationID) error { | |||
| 1240 | 1240 | // finishSnapshotPurge removes catalogue metadata first, then deletes the | |
| 1241 | 1241 | // incarnation directory. Startup garbage collection removes the directory if | |
| 1242 | 1242 | // the second step is interrupted. | |
| 1243 | - | func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, _ domain.IncarnationID) error { | |
| 1243 | + | func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error { | |
| 1244 | 1244 | if !d.persistEnabled { | |
| 1245 | 1245 | return nil | |
| 1246 | 1246 | } | |
| @@ -1252,7 +1252,7 @@ func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, _ domain. | |||
| 1252 | 1252 | } | |
| 1253 | 1253 | purgeCtx, cancel := context.WithTimeout(ctx, snapshotFinalFlushTimeout) | |
| 1254 | 1254 | defer cancel() | |
| 1255 | - | return d.recovery.Delete(purgeCtx, name) | |
| 1255 | + | return d.recovery.DeleteExact(purgeCtx, name, incarnation, createdAt) | |
| 1256 | 1256 | } | |
| 1257 | 1257 | ||
| 1258 | 1258 | // retryStoppedPurge completes a previously closed session's durable purge. | |
| @@ -1285,16 +1285,16 @@ func (d *Daemon) retryStoppedPurgeContextExact(ctx context.Context, name string, | |||
| 1285 | 1285 | } | |
| 1286 | 1286 | stopped.purging = true | |
| 1287 | 1287 | d.inactive[name] = stopped | |
| 1288 | + | d.mu.Unlock() | |
| 1288 | 1289 | ||
| 1289 | 1290 | if err := d.beginSnapshotPurge(name, stopped.incarnation); err != nil { | |
| 1290 | - | d.mu.Unlock() | |
| 1291 | 1291 | return err | |
| 1292 | 1292 | } | |
| 1293 | - | if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation); err != nil { | |
| 1294 | - | d.mu.Unlock() | |
| 1293 | + | if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation, stopped.createdAt); err != nil { | |
| 1295 | 1294 | return err | |
| 1296 | 1295 | } | |
| 1297 | - | if current, ok := d.inactive[name]; ok && current.purging { | |
| 1296 | + | d.mu.Lock() | |
| 1297 | + | if current, ok := d.inactive[name]; ok && current.purging && current.sameLifecycle(stopped) { | |
| 1298 | 1298 | delete(d.inactive, name) | |
| 1299 | 1299 | } | |
| 1300 | 1300 | d.mu.Unlock() | |
| @@ -1810,17 +1810,18 @@ func (d *Daemon) killSessionWithSnapshotDeadlineAndCondition(sess *session, reas | |||
| 1810 | 1810 | // another generation after this destructive source deletion. Keep the hidden | |
| 1811 | 1811 | // stopped record when deletion fails so a repeated live kill can retry. | |
| 1812 | 1812 | if !ephemeral && purge { | |
| 1813 | - | d.mu.Lock() | |
| 1814 | - | sess.mu.Lock() | |
| 1815 | - | err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation) | |
| 1816 | - | sess.mu.Unlock() | |
| 1813 | + | err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation, stoppedRecord.CreatedAt) | |
| 1817 | 1814 | if err != nil { | |
| 1818 | 1815 | purgeErr = errors.Join(purgeErr, err) | |
| 1819 | 1816 | d.log.Warn("finishing session snapshot purge failed", "err", err, "session", stoppedName) | |
| 1820 | - | } else if stopped, ok := d.inactive[stoppedName]; ok && stopped.purging { | |
| 1821 | - | delete(d.inactive, stoppedName) | |
| 1817 | + | } else { | |
| 1818 | + | expected := inactiveSessionFromRecord(stoppedRecord, protocol.SessionDown, nil) | |
| 1819 | + | d.mu.Lock() | |
| 1820 | + | if stopped, ok := d.inactive[stoppedName]; ok && stopped.purging && stopped.sameLifecycle(expected) { | |
| 1821 | + | delete(d.inactive, stoppedName) | |
| 1822 | + | } | |
| 1823 | + | d.mu.Unlock() | |
| 1822 | 1824 | } | |
| 1823 | - | d.mu.Unlock() | |
| 1824 | 1825 | } | |
| 1825 | 1826 | if empty { | |
| 1826 | 1827 | d.doneOnce.Do(func() { close(d.done) }) | |
| @@ -1165,9 +1165,9 @@ func (d *Daemon) closeTabLockedWithEffect(sess *session, tb *tab, repaint bool, | |||||
| 1165 | var err error | 1165 | var err error | ||
| 1166 | if effect != nil { | 1166 | if effect != nil { | ||
| 1167 | effect.bindActionEnd(d, "close-tab") | 1167 | effect.bindActionEnd(d, "close-tab") | ||
| 1168 | - | err = d.killSessionForAttachment(sess, protocol.ReasonSessionKilled, false, effect, "close-tab") | 1168 | + | err = d.killSessionForAttachment(sess, protocol.ReasonSessionKilled, true, effect, "close-tab") |
| 1169 | } else { | 1169 | } else { | ||
| 1170 | - | err = d.killSession(sess, protocol.ReasonSessionKilled, false) | 1170 | + | err = d.killSession(sess, protocol.ReasonSessionKilled, true) |
| 1171 | } | 1171 | } | ||
| 1172 | if err != nil { | 1172 | if err != nil { | ||
| 1173 | d.log.Warn("closing last tab failed", "session", name, "err", err) | 1173 | d.log.Warn("closing last tab failed", "session", name, "err", err) | ||
| @@ -1240,7 +1240,7 @@ func (d *Daemon) beginSnapshotPurge(_ string, _ domain.IncarnationID) error { | |||||
| 1240 | // finishSnapshotPurge removes catalogue metadata first, then deletes the | 1240 | // finishSnapshotPurge removes catalogue metadata first, then deletes the | ||
| 1241 | // incarnation directory. Startup garbage collection removes the directory if | 1241 | // incarnation directory. Startup garbage collection removes the directory if | ||
| 1242 | // the second step is interrupted. | 1242 | // the second step is interrupted. | ||
| 1243 | - | func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, _ domain.IncarnationID) error { | 1243 | + | func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error { |
| 1244 | if !d.persistEnabled { | 1244 | if !d.persistEnabled { | ||
| 1245 | return nil | 1245 | return nil | ||
| 1246 | } | 1246 | } | ||
| @@ -1252,7 +1252,7 @@ func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, _ domain. | |||||
| 1252 | } | 1252 | } | ||
| 1253 | purgeCtx, cancel := context.WithTimeout(ctx, snapshotFinalFlushTimeout) | 1253 | purgeCtx, cancel := context.WithTimeout(ctx, snapshotFinalFlushTimeout) | ||
| 1254 | defer cancel() | 1254 | defer cancel() | ||
| 1255 | - | return d.recovery.Delete(purgeCtx, name) | 1255 | + | return d.recovery.DeleteExact(purgeCtx, name, incarnation, createdAt) |
| 1256 | } | 1256 | } | ||
| 1257 | 1257 | ||||
| 1258 | // retryStoppedPurge completes a previously closed session's durable purge. | 1258 | // retryStoppedPurge completes a previously closed session's durable purge. | ||
| @@ -1285,16 +1285,16 @@ func (d *Daemon) retryStoppedPurgeContextExact(ctx context.Context, name string, | |||||
| 1285 | } | 1285 | } | ||
| 1286 | stopped.purging = true | 1286 | stopped.purging = true | ||
| 1287 | d.inactive[name] = stopped | 1287 | d.inactive[name] = stopped | ||
| 1288 | + | d.mu.Unlock() | |||
| 1288 | 1289 | ||||
| 1289 | if err := d.beginSnapshotPurge(name, stopped.incarnation); err != nil { | 1290 | if err := d.beginSnapshotPurge(name, stopped.incarnation); err != nil { | ||
| 1290 | - | d.mu.Unlock() | |||
| 1291 | return err | 1291 | return err | ||
| 1292 | } | 1292 | } | ||
| 1293 | - | if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation); err != nil { | 1293 | + | if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation, stopped.createdAt); err != nil { |
| 1294 | - | d.mu.Unlock() | |||
| 1295 | return err | 1294 | return err | ||
| 1296 | } | 1295 | } | ||
| 1297 | - | if current, ok := d.inactive[name]; ok && current.purging { | 1296 | + | d.mu.Lock() |
| 1297 | + | if current, ok := d.inactive[name]; ok && current.purging && current.sameLifecycle(stopped) { | |||
| 1298 | delete(d.inactive, name) | 1298 | delete(d.inactive, name) | ||
| 1299 | } | 1299 | } | ||
| 1300 | d.mu.Unlock() | 1300 | d.mu.Unlock() | ||
| @@ -1810,17 +1810,18 @@ func (d *Daemon) killSessionWithSnapshotDeadlineAndCondition(sess *session, reas | |||||
| 1810 | // another generation after this destructive source deletion. Keep the hidden | 1810 | // another generation after this destructive source deletion. Keep the hidden | ||
| 1811 | // stopped record when deletion fails so a repeated live kill can retry. | 1811 | // stopped record when deletion fails so a repeated live kill can retry. | ||
| 1812 | if !ephemeral && purge { | 1812 | if !ephemeral && purge { | ||
| 1813 | - | d.mu.Lock() | 1813 | + | err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation, stoppedRecord.CreatedAt) |
| 1814 | - | sess.mu.Lock() | |||
| 1815 | - | err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation) | |||
| 1816 | - | sess.mu.Unlock() | |||
| 1817 | if err != nil { | 1814 | if err != nil { | ||
| 1818 | purgeErr = errors.Join(purgeErr, err) | 1815 | purgeErr = errors.Join(purgeErr, err) | ||
| 1819 | d.log.Warn("finishing session snapshot purge failed", "err", err, "session", stoppedName) | 1816 | d.log.Warn("finishing session snapshot purge failed", "err", err, "session", stoppedName) | ||
| 1820 | - | } else if stopped, ok := d.inactive[stoppedName]; ok && stopped.purging { | 1817 | + | } else { |
| 1821 | - | delete(d.inactive, stoppedName) | 1818 | + | expected := inactiveSessionFromRecord(stoppedRecord, protocol.SessionDown, nil) |
| 1819 | + | d.mu.Lock() | |||
| 1820 | + | if stopped, ok := d.inactive[stoppedName]; ok && stopped.purging && stopped.sameLifecycle(expected) { | |||
| 1821 | + | delete(d.inactive, stoppedName) | |||
| 1822 | + | } | |||
| 1823 | + | d.mu.Unlock() | |||
| 1822 | } | 1824 | } | ||
| 1823 | - | d.mu.Unlock() | |||
| 1824 | } | 1825 | } | ||
| 1825 | if empty { | 1826 | if empty { | ||
| 1826 | d.doneOnce.Do(func() { close(d.done) }) | 1827 | d.doneOnce.Do(func() { close(d.done) }) | ||
M internal/usecase/daemon/session_test.go +1 -1
| @@ -1232,7 +1232,7 @@ func TestPickerStoppedTargetKillPurges(t *testing.T) { | |||
| 1232 | 1232 | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 1233 | 1233 | WithStore(t, store)(d) | |
| 1234 | 1234 | require.NoError(t, testPersister(t, d).Save(persist.Record{Name: "old", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 1})) | |
| 1235 | - | d.inactive["old"] = inactiveSession{name: "old", cwd: "/tmp", createdAt: 1} | |
| 1235 | + | d.inactive["old"] = inactiveSession{name: "old", cwd: "/tmp", createdAt: 1, incarnation: domain.IncarnationID{1}} | |
| 1236 | 1236 | require.NoError(t, d.killPickerTarget(picker.Target{Name: "old", Stopped: true})) | |
| 1237 | 1237 | require.False(t, state.has("old")) | |
| 1238 | 1238 | d.mu.Lock() | |
| @@ -1232,7 +1232,7 @@ func TestPickerStoppedTargetKillPurges(t *testing.T) { | |||||
| 1232 | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | 1232 | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | ||
| 1233 | WithStore(t, store)(d) | 1233 | WithStore(t, store)(d) | ||
| 1234 | require.NoError(t, testPersister(t, d).Save(persist.Record{Name: "old", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 1})) | 1234 | require.NoError(t, testPersister(t, d).Save(persist.Record{Name: "old", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 1})) | ||
| 1235 | - | d.inactive["old"] = inactiveSession{name: "old", cwd: "/tmp", createdAt: 1} | 1235 | + | d.inactive["old"] = inactiveSession{name: "old", cwd: "/tmp", createdAt: 1, incarnation: domain.IncarnationID{1}} |
| 1236 | require.NoError(t, d.killPickerTarget(picker.Target{Name: "old", Stopped: true})) | 1236 | require.NoError(t, d.killPickerTarget(picker.Target{Name: "old", Stopped: true})) | ||
| 1237 | require.False(t, state.has("old")) | 1237 | require.False(t, state.has("old")) | ||
| 1238 | d.mu.Lock() | 1238 | d.mu.Lock() | ||
M internal/usecase/daemon/snapshot_scheduler_test.go +6 -5
| @@ -287,8 +287,9 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing. | |||
| 287 | 287 | } | |
| 288 | 288 | } | |
| 289 | 289 | ||
| 290 | - | func TestServeShutdownDeadlineStillJoinsPaneExitWriter(t *testing.T) { | |
| 290 | + | func TestServeShutdownDeadlineStillJoinsRetainedSessionWriter(t *testing.T) { | |
| 291 | 291 | pty, releasePTY := newBlockingPTY(t) | |
| 292 | + | defer releasePTY() | |
| 292 | 293 | clock := newServeShutdownClock() | |
| 293 | 294 | d := newTestDaemon(t, newFactory(t, pty), clock) | |
| 294 | 295 | repository := portsmocks.NewMockSnapshotRepository(t) | |
| @@ -316,10 +317,10 @@ func TestServeShutdownDeadlineStillJoinsPaneExitWriter(t *testing.T) { | |||
| 316 | 317 | sess := firstSession(d) | |
| 317 | 318 | require.NotNil(t, sess) | |
| 318 | 319 | ||
| 319 | - | // Releasing the real pane reader drives EOF through ptyReader, reapPane, and | |
| 320 | - | // closePane. That ordinary lifecycle path must own teardown before Serve's | |
| 321 | - | // context cancellation enters its competing shutdownAll path. | |
| 322 | - | releasePTY() | |
| 320 | + | // A retained-session teardown still publishes a terminal snapshot. It must | |
| 321 | + | // own teardown before Serve's context cancellation enters its competing | |
| 322 | + | // shutdownAll path. | |
| 323 | + | go func() { _ = d.killSession(sess, protocol.ReasonServerShutdown, false) }() | |
| 323 | 324 | select { | |
| 324 | 325 | case <-publicationStarted: | |
| 325 | 326 | case <-time.After(testWaitTimeout): | |
| @@ -287,8 +287,9 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing. | |||||
| 287 | } | 287 | } | ||
| 288 | } | 288 | } | ||
| 289 | 289 | ||||
| 290 | - | func TestServeShutdownDeadlineStillJoinsPaneExitWriter(t *testing.T) { | 290 | + | func TestServeShutdownDeadlineStillJoinsRetainedSessionWriter(t *testing.T) { |
| 291 | pty, releasePTY := newBlockingPTY(t) | 291 | pty, releasePTY := newBlockingPTY(t) | ||
| 292 | + | defer releasePTY() | |||
| 292 | clock := newServeShutdownClock() | 293 | clock := newServeShutdownClock() | ||
| 293 | d := newTestDaemon(t, newFactory(t, pty), clock) | 294 | d := newTestDaemon(t, newFactory(t, pty), clock) | ||
| 294 | repository := portsmocks.NewMockSnapshotRepository(t) | 295 | repository := portsmocks.NewMockSnapshotRepository(t) | ||
| @@ -316,10 +317,10 @@ func TestServeShutdownDeadlineStillJoinsPaneExitWriter(t *testing.T) { | |||||
| 316 | sess := firstSession(d) | 317 | sess := firstSession(d) | ||
| 317 | require.NotNil(t, sess) | 318 | require.NotNil(t, sess) | ||
| 318 | 319 | ||||
| 319 | - | // Releasing the real pane reader drives EOF through ptyReader, reapPane, and | 320 | + | // A retained-session teardown still publishes a terminal snapshot. It must |
| 320 | - | // closePane. That ordinary lifecycle path must own teardown before Serve's | 321 | + | // own teardown before Serve's context cancellation enters its competing |
| 321 | - | // context cancellation enters its competing shutdownAll path. | 322 | + | // shutdownAll path. |
| 322 | - | releasePTY() | 323 | + | go func() { _ = d.killSession(sess, protocol.ReasonServerShutdown, false) }() |
| 323 | select { | 324 | select { | ||
| 324 | case <-publicationStarted: | 325 | case <-publicationStarted: | ||
| 325 | case <-time.After(testWaitTimeout): | 326 | case <-time.After(testWaitTimeout): | ||
M internal/usecase/recovery/coordinator.go +29 -2
| @@ -170,8 +170,8 @@ func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (doma | |||
| 170 | 170 | return record, nil | |
| 171 | 171 | } | |
| 172 | 172 | ||
| 173 | - | // Delete removes the catalogue record before deleting its incarnation. A crash | |
| 174 | - | // between those steps leaves an orphan that startup garbage collection removes. | |
| 173 | + | // Delete removes the current catalogue record before deleting its incarnation. | |
| 174 | + | // Callers that captured lifecycle identity must use DeleteExact. | |
| 175 | 175 | func (c *Coordinator) Delete(ctx context.Context, name string) error { | |
| 176 | 176 | if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil { | |
| 177 | 177 | return errors.New("recovery: incomplete delete dependencies") | |
| @@ -196,6 +196,33 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error { | |||
| 196 | 196 | return c.repository.DeleteIncarnation(ctx, record.IncarnationID) | |
| 197 | 197 | } | |
| 198 | 198 | ||
| 199 | + | // DeleteExact removes a catalogue record only while it still identifies the | |
| 200 | + | // lifecycle captured by the caller. A stale teardown cannot delete a same-name | |
| 201 | + | // replacement created after that capture. | |
| 202 | + | func (c *Coordinator) DeleteExact(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error { | |
| 203 | + | if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil { | |
| 204 | + | return errors.New("recovery: incomplete exact delete dependencies") | |
| 205 | + | } | |
| 206 | + | if err := ctx.Err(); err != nil { | |
| 207 | + | return err | |
| 208 | + | } | |
| 209 | + | c.mutationMu.Lock() | |
| 210 | + | defer c.mutationMu.Unlock() | |
| 211 | + | unlock := c.locks.Lock([]string{name}) | |
| 212 | + | defer unlock() | |
| 213 | + | record, ok, err := c.catalogue.Record(name) | |
| 214 | + | if err != nil { | |
| 215 | + | return err | |
| 216 | + | } | |
| 217 | + | if !ok || record.IncarnationID != incarnation || record.CreatedAt != createdAt { | |
| 218 | + | return nil | |
| 219 | + | } | |
| 220 | + | if err := c.catalogue.Delete(name); err != nil { | |
| 221 | + | return err | |
| 222 | + | } | |
| 223 | + | return c.repository.DeleteIncarnation(ctx, incarnation) | |
| 224 | + | } | |
| 225 | + | ||
| 199 | 226 | // CollectGarbage takes a catalogue snapshot and applies retention while all | |
| 200 | 227 | // catalogue mutations and checkpoint publications are fenced. Its keep map is | |
| 201 | 228 | // therefore never stale relative to an operation that creates an incarnation | |
| @@ -170,8 +170,8 @@ func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (doma | |||||
| 170 | return record, nil | 170 | return record, nil | ||
| 171 | } | 171 | } | ||
| 172 | 172 | ||||
| 173 | - | // Delete removes the catalogue record before deleting its incarnation. A crash | 173 | + | // Delete removes the current catalogue record before deleting its incarnation. |
| 174 | - | // between those steps leaves an orphan that startup garbage collection removes. | 174 | + | // Callers that captured lifecycle identity must use DeleteExact. |
| 175 | func (c *Coordinator) Delete(ctx context.Context, name string) error { | 175 | func (c *Coordinator) Delete(ctx context.Context, name string) error { | ||
| 176 | if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil { | 176 | if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil { | ||
| 177 | return errors.New("recovery: incomplete delete dependencies") | 177 | return errors.New("recovery: incomplete delete dependencies") | ||
| @@ -196,6 +196,33 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error { | |||||
| 196 | return c.repository.DeleteIncarnation(ctx, record.IncarnationID) | 196 | return c.repository.DeleteIncarnation(ctx, record.IncarnationID) | ||
| 197 | } | 197 | } | ||
| 198 | 198 | ||||
| 199 | + | // DeleteExact removes a catalogue record only while it still identifies the | |||
| 200 | + | // lifecycle captured by the caller. A stale teardown cannot delete a same-name | |||
| 201 | + | // replacement created after that capture. | |||
| 202 | + | func (c *Coordinator) DeleteExact(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error { | |||
| 203 | + | if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil { | |||
| 204 | + | return errors.New("recovery: incomplete exact delete dependencies") | |||
| 205 | + | } | |||
| 206 | + | if err := ctx.Err(); err != nil { | |||
| 207 | + | return err | |||
| 208 | + | } | |||
| 209 | + | c.mutationMu.Lock() | |||
| 210 | + | defer c.mutationMu.Unlock() | |||
| 211 | + | unlock := c.locks.Lock([]string{name}) | |||
| 212 | + | defer unlock() | |||
| 213 | + | record, ok, err := c.catalogue.Record(name) | |||
| 214 | + | if err != nil { | |||
| 215 | + | return err | |||
| 216 | + | } | |||
| 217 | + | if !ok || record.IncarnationID != incarnation || record.CreatedAt != createdAt { | |||
| 218 | + | return nil | |||
| 219 | + | } | |||
| 220 | + | if err := c.catalogue.Delete(name); err != nil { | |||
| 221 | + | return err | |||
| 222 | + | } | |||
| 223 | + | return c.repository.DeleteIncarnation(ctx, incarnation) | |||
| 224 | + | } | |||
| 225 | + | ||||
| 199 | // CollectGarbage takes a catalogue snapshot and applies retention while all | 226 | // CollectGarbage takes a catalogue snapshot and applies retention while all | ||
| 200 | // catalogue mutations and checkpoint publications are fenced. Its keep map is | 227 | // catalogue mutations and checkpoint publications are fenced. Its keep map is | ||
| 201 | // therefore never stale relative to an operation that creates an incarnation | 228 | // therefore never stale relative to an operation that creates an incarnation | ||
M internal/usecase/recovery/transactions_test.go +28 -0
| @@ -294,6 +294,34 @@ func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) { | |||
| 294 | 294 | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs) | |
| 295 | 295 | } | |
| 296 | 296 | ||
| 297 | + | func TestDeleteExactDoesNotDeleteSameNameReplacement(t *testing.T) { | |
| 298 | + | oldID := domain.IncarnationID{1} | |
| 299 | + | newID := domain.IncarnationID{2} | |
| 300 | + | catalogue := newTransactionCatalogue(domain.CatalogueRecord{Name: "work", IncarnationID: newID, Cwd: "/tmp", CreatedAt: 22, UpdatedAt: 22}) | |
| 301 | + | repository := &transactionRepository{} | |
| 302 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil)) | |
| 303 | + | ||
| 304 | + | require.NoError(t, coordinator.DeleteExact(context.Background(), "work", oldID, 11)) | |
| 305 | + | record, ok, err := catalogue.Record("work") | |
| 306 | + | require.NoError(t, err) | |
| 307 | + | require.True(t, ok) | |
| 308 | + | require.Equal(t, newID, record.IncarnationID) | |
| 309 | + | require.Empty(t, repository.deleteCalls) | |
| 310 | + | } | |
| 311 | + | ||
| 312 | + | func TestDeleteExactDeletesMatchingLifecycle(t *testing.T) { | |
| 313 | + | id := domain.IncarnationID{1} | |
| 314 | + | catalogue := newTransactionCatalogue(domain.CatalogueRecord{Name: "work", IncarnationID: id, Cwd: "/tmp", CreatedAt: 11, UpdatedAt: 11}) | |
| 315 | + | repository := &transactionRepository{} | |
| 316 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil)) | |
| 317 | + | ||
| 318 | + | require.NoError(t, coordinator.DeleteExact(context.Background(), "work", id, 11)) | |
| 319 | + | _, ok, err := catalogue.Record("work") | |
| 320 | + | require.NoError(t, err) | |
| 321 | + | require.False(t, ok) | |
| 322 | + | require.Equal(t, []domain.IncarnationID{id}, repository.deleteCalls) | |
| 323 | + | } | |
| 324 | + | ||
| 297 | 325 | func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) { | |
| 298 | 326 | t.Parallel() | |
| 299 | 327 | old := healthyTransactionRecord() | |
| @@ -294,6 +294,34 @@ func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) { | |||||
| 294 | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs) | 294 | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs) | ||
| 295 | } | 295 | } | ||
| 296 | 296 | ||||
| 297 | + | func TestDeleteExactDoesNotDeleteSameNameReplacement(t *testing.T) { | |||
| 298 | + | oldID := domain.IncarnationID{1} | |||
| 299 | + | newID := domain.IncarnationID{2} | |||
| 300 | + | catalogue := newTransactionCatalogue(domain.CatalogueRecord{Name: "work", IncarnationID: newID, Cwd: "/tmp", CreatedAt: 22, UpdatedAt: 22}) | |||
| 301 | + | repository := &transactionRepository{} | |||
| 302 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil)) | |||
| 303 | + | ||||
| 304 | + | require.NoError(t, coordinator.DeleteExact(context.Background(), "work", oldID, 11)) | |||
| 305 | + | record, ok, err := catalogue.Record("work") | |||
| 306 | + | require.NoError(t, err) | |||
| 307 | + | require.True(t, ok) | |||
| 308 | + | require.Equal(t, newID, record.IncarnationID) | |||
| 309 | + | require.Empty(t, repository.deleteCalls) | |||
| 310 | + | } | |||
| 311 | + | ||||
| 312 | + | func TestDeleteExactDeletesMatchingLifecycle(t *testing.T) { | |||
| 313 | + | id := domain.IncarnationID{1} | |||
| 314 | + | catalogue := newTransactionCatalogue(domain.CatalogueRecord{Name: "work", IncarnationID: id, Cwd: "/tmp", CreatedAt: 11, UpdatedAt: 11}) | |||
| 315 | + | repository := &transactionRepository{} | |||
| 316 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil)) | |||
| 317 | + | ||||
| 318 | + | require.NoError(t, coordinator.DeleteExact(context.Background(), "work", id, 11)) | |||
| 319 | + | _, ok, err := catalogue.Record("work") | |||
| 320 | + | require.NoError(t, err) | |||
| 321 | + | require.False(t, ok) | |||
| 322 | + | require.Equal(t, []domain.IncarnationID{id}, repository.deleteCalls) | |||
| 323 | + | } | |||
| 324 | + | ||||
| 297 | func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) { | 325 | func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) { | ||
| 298 | t.Parallel() | 326 | t.Parallel() | ||
| 299 | old := healthyTransactionRecord() | 327 | old := healthyTransactionRecord() | ||