Merge pull request 'refactor(client): split Supervisor state into navigation and picker owners' (#11) from refactor/supervisor-split into main
9 files · +250 -110
M internal/usecase/client/preview_manager_test.go +5 -5
| @@ -342,15 +342,15 @@ func TestSupervisorRefreshPreviewWiresTerminalGeometryToPreviewManager(t *testin | |||
| 342 | 342 | h.service.EXPECT().SubscribePreview(expected).Return(sub, nil).Once() | |
| 343 | 343 | sub.EXPECT().Close().Once() | |
| 344 | 344 | ||
| 345 | - | supervisor := &Supervisor{cfg: SupervisorConfig{Picker: h.controller, Terminal: &pickerRecordingTerminal{}}, preview: previewManager{clock: h.clock}} | |
| 345 | + | supervisor := &Supervisor{cfg: SupervisorConfig{Picker: h.controller, Terminal: &pickerRecordingTerminal{}}, picker: pickerSession{preview: previewManager{clock: h.clock}}} | |
| 346 | 346 | supervisor.refreshPreview(h.service) | |
| 347 | 347 | h.clock.awaitTimer(t).fire() | |
| 348 | 348 | select { | |
| 349 | - | case <-supervisor.preview.changed(): | |
| 349 | + | case <-supervisor.picker.preview.changed(): | |
| 350 | 350 | case <-time.After(5 * time.Second): | |
| 351 | 351 | t.Fatal("the supervisor preview did not wake") | |
| 352 | 352 | } | |
| 353 | - | supervisor.preview.publish(h.controller) | |
| 354 | - | require.Equal(t, expected, supervisor.preview.request) | |
| 355 | - | supervisor.preview.close(h.controller) | |
| 353 | + | supervisor.picker.preview.publish(h.controller) | |
| 354 | + | require.Equal(t, expected, supervisor.picker.preview.request) | |
| 355 | + | supervisor.picker.preview.close(h.controller) | |
| 356 | 356 | } | |
| @@ -342,15 +342,15 @@ func TestSupervisorRefreshPreviewWiresTerminalGeometryToPreviewManager(t *testin | |||||
| 342 | h.service.EXPECT().SubscribePreview(expected).Return(sub, nil).Once() | 342 | h.service.EXPECT().SubscribePreview(expected).Return(sub, nil).Once() | ||
| 343 | sub.EXPECT().Close().Once() | 343 | sub.EXPECT().Close().Once() | ||
| 344 | 344 | ||||
| 345 | - | supervisor := &Supervisor{cfg: SupervisorConfig{Picker: h.controller, Terminal: &pickerRecordingTerminal{}}, preview: previewManager{clock: h.clock}} | 345 | + | supervisor := &Supervisor{cfg: SupervisorConfig{Picker: h.controller, Terminal: &pickerRecordingTerminal{}}, picker: pickerSession{preview: previewManager{clock: h.clock}}} |
| 346 | supervisor.refreshPreview(h.service) | 346 | supervisor.refreshPreview(h.service) | ||
| 347 | h.clock.awaitTimer(t).fire() | 347 | h.clock.awaitTimer(t).fire() | ||
| 348 | select { | 348 | select { | ||
| 349 | - | case <-supervisor.preview.changed(): | 349 | + | case <-supervisor.picker.preview.changed(): |
| 350 | case <-time.After(5 * time.Second): | 350 | case <-time.After(5 * time.Second): | ||
| 351 | t.Fatal("the supervisor preview did not wake") | 351 | t.Fatal("the supervisor preview did not wake") | ||
| 352 | } | 352 | } | ||
| 353 | - | supervisor.preview.publish(h.controller) | 353 | + | supervisor.picker.preview.publish(h.controller) |
| 354 | - | require.Equal(t, expected, supervisor.preview.request) | 354 | + | require.Equal(t, expected, supervisor.picker.preview.request) |
| 355 | - | supervisor.preview.close(h.controller) | 355 | + | supervisor.picker.preview.close(h.controller) |
| 356 | } | 356 | } | ||
M internal/usecase/client/supervisor.go +20 -40
| @@ -434,10 +434,9 @@ type Supervisor struct { | |||
| 434 | 434 | cfg SupervisorConfig | |
| 435 | 435 | logger *slog.Logger | |
| 436 | 436 | ||
| 437 | - | // pendingPickerKey retains a commit observed while connecting. It is | |
| 438 | - | // revalidated against the adopted service just like a ready-phase commit. | |
| 439 | - | pendingPickerKey string | |
| 440 | - | preview previewManager | |
| 437 | + | // picker groups the picker's per-run state: the pending commit key, the | |
| 438 | + | // preview subscription, and the off-goroutine kill runner. | |
| 439 | + | picker pickerSession | |
| 441 | 440 | ||
| 442 | 441 | mu sync.Mutex | |
| 443 | 442 | state State | |
| @@ -468,28 +467,9 @@ type Supervisor struct { | |||
| 468 | 467 | // capabilities is the outer terminal's probed and enabled capabilities, | |
| 469 | 468 | // written once in Run before any attachment starts. | |
| 470 | 469 | capabilities terminalCapabilities | |
| 471 | - | // readySub is the adopted connection's subscription while the ready phase | |
| 472 | - | // runs, so the picker overlay over a live attachment keeps folding broker | |
| 473 | - | // publications. It is only touched from the run goroutine. | |
| 474 | - | readySub ports.BrokerSubscription | |
| 475 | - | // pendingSwap is the request the picker overlay or a daemon navigation | |
| 476 | - | // committed to another target while an attachment was live. It is only | |
| 477 | - | // touched from the run goroutine and consumed by settleAttachment. | |
| 478 | - | pendingSwap *pickerAttachmentTarget | |
| 479 | - | // pendingInPlace is the choice the live attachment is switching to in | |
| 480 | - | // place on its own daemon. A refusal turns it into pendingSwap. Only | |
| 481 | - | // touched from the run goroutine; cleared when the attachment settles. | |
| 482 | - | pendingInPlace *pendingInPlace | |
| 483 | - | // inPlaceSeq numbers in-place choices so a late outcome for an older one | |
| 484 | - | // is never applied to a newer one. | |
| 485 | - | inPlaceSeq uint64 | |
| 486 | - | // kills runs the picker's `x` operations off the run goroutine. | |
| 487 | - | kills pickerKills | |
| 488 | - | // routes is the client route ledger published to the serving daemon; | |
| 489 | - | // routesSent is the attachment that received its latest snapshot. Both are | |
| 490 | - | // only touched from the run goroutine. | |
| 491 | - | routes *routeLedger | |
| 492 | - | routesSent AttachmentToken | |
| 470 | + | // nav is the run goroutine's navigation bookkeeping: the ready-phase | |
| 471 | + | // subscription, pending swap and in-place choices, and the route ledger. | |
| 472 | + | nav navigationState | |
| 493 | 473 | } | |
| 494 | 474 | ||
| 495 | 475 | // NewSupervisor validates the required dependencies and returns a supervisor | |
| @@ -537,7 +517,7 @@ func NewSupervisor(cfg SupervisorConfig) (*Supervisor, error) { | |||
| 537 | 517 | state: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected}, | |
| 538 | 518 | clientID: newClientID(), | |
| 539 | 519 | navigation: cfg.InitialNavigation, | |
| 540 | - | preview: previewManager{clock: cfg.Clock}, | |
| 520 | + | picker: pickerSession{preview: previewManager{clock: cfg.Clock}}, | |
| 541 | 521 | } | |
| 542 | 522 | // The host is the supervisor's existing foreground grant. It owns no raw | |
| 543 | 523 | // mode and starts no reader: the supervisor keeps its one terminal input | |
| @@ -609,8 +589,8 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||
| 609 | 589 | ) | |
| 610 | 590 | retire := func() { | |
| 611 | 591 | s.retirePickerKill() | |
| 612 | - | s.preview.close(s.cfg.Picker) | |
| 613 | - | s.readySub = nil | |
| 592 | + | s.picker.preview.close(s.cfg.Picker) | |
| 593 | + | s.nav.readySub = nil | |
| 614 | 594 | if sub != nil { | |
| 615 | 595 | sub.Close() | |
| 616 | 596 | sub = nil | |
| @@ -679,7 +659,7 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||
| 679 | 659 | } | |
| 680 | 660 | ||
| 681 | 661 | service, sub = result.service, result.sub | |
| 682 | - | s.readySub = sub | |
| 662 | + | s.nav.readySub = sub | |
| 683 | 663 | if s.cfg.Picker != nil { | |
| 684 | 664 | // Render the first committed publication immediately, before waiting | |
| 685 | 665 | // for the next one; the picker never shows a stale empty catalogue | |
| @@ -915,8 +895,8 @@ type readyOutcome struct { | |||
| 915 | 895 | // the supervisor can admit exactly the row the user committed. It opens no | |
| 916 | 896 | // stream and starts no terminal work itself. | |
| 917 | 897 | func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, sub ports.BrokerSubscription) readyOutcome { | |
| 918 | - | if key := s.pendingPickerKey; key != "" { | |
| 919 | - | s.pendingPickerKey = "" | |
| 898 | + | if key := s.picker.pendingPickerKey; key != "" { | |
| 899 | + | s.picker.pendingPickerKey = "" | |
| 920 | 900 | return readyOutcome{commitKey: key} | |
| 921 | 901 | } | |
| 922 | 902 | var changed <-chan struct{} | |
| @@ -939,11 +919,11 @@ func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetim | |||
| 939 | 919 | s.cfg.Picker.ApplySnapshot(service.Snapshot()) | |
| 940 | 920 | s.refreshPreview(service) | |
| 941 | 921 | s.renderCurrent() | |
| 942 | - | case <-s.preview.changed(): | |
| 943 | - | if s.preview.publish(s.cfg.Picker) { | |
| 922 | + | case <-s.picker.preview.changed(): | |
| 923 | + | if s.picker.preview.publish(s.cfg.Picker) { | |
| 944 | 924 | s.renderCurrent() | |
| 945 | 925 | } | |
| 946 | - | case outcome := <-s.kills.results(): | |
| 926 | + | case outcome := <-s.picker.kills.results(): | |
| 947 | 927 | s.finishPickerKill(service, outcome) | |
| 948 | 928 | case <-ops: | |
| 949 | 929 | op, key := s.cfg.Picker.TakeOp() | |
| @@ -961,7 +941,7 @@ func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetim | |||
| 961 | 941 | if op.exit { | |
| 962 | 942 | // Ctrl+C is the explicit exit. With no attachment to | |
| 963 | 943 | // return to, it is the only close that ends the process. | |
| 964 | - | s.preview.close(s.cfg.Picker) | |
| 944 | + | s.picker.preview.close(s.cfg.Picker) | |
| 965 | 945 | return readyOutcome{terminated: true} | |
| 966 | 946 | } | |
| 967 | 947 | // Escape and q cancel; with no attachment there is nothing to | |
| @@ -971,7 +951,7 @@ func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetim | |||
| 971 | 951 | continue | |
| 972 | 952 | } | |
| 973 | 953 | if op.commit && key != "" { | |
| 974 | - | s.preview.close(s.cfg.Picker) | |
| 954 | + | s.picker.preview.close(s.cfg.Picker) | |
| 975 | 955 | return readyOutcome{commitKey: key} | |
| 976 | 956 | } | |
| 977 | 957 | case <-ctx.Done(): | |
| @@ -991,10 +971,10 @@ func (s *Supervisor) refreshPreview(service ports.BrokerNavigator) { | |||
| 991 | 971 | } | |
| 992 | 972 | geometry, err := s.cfg.Terminal.Geometry() | |
| 993 | 973 | if err != nil || !geometry.Valid() { | |
| 994 | - | s.preview.close(s.cfg.Picker) | |
| 974 | + | s.picker.preview.close(s.cfg.Picker) | |
| 995 | 975 | return | |
| 996 | 976 | } | |
| 997 | - | s.preview.refresh(service, s.cfg.Picker, geometry.Size) | |
| 977 | + | s.picker.preview.refresh(service, s.cfg.Picker, geometry.Size) | |
| 998 | 978 | } | |
| 999 | 979 | ||
| 1000 | 980 | // pickerOps is nil when no picker owns presentation decisions. | |
| @@ -1011,7 +991,7 @@ func (s *Supervisor) pickerOps() <-chan struct{} { | |||
| 1011 | 991 | func (s *Supervisor) takePickerClose() bool { | |
| 1012 | 992 | op, key := s.cfg.Picker.TakeOp() | |
| 1013 | 993 | if op.commit && key != "" { | |
| 1014 | - | s.pendingPickerKey = key | |
| 994 | + | s.picker.pendingPickerKey = key | |
| 1015 | 995 | } | |
| 1016 | 996 | if op.close && !op.exit { | |
| 1017 | 997 | s.notifyPicker(pickerExitHint) | |
| @@ -434,10 +434,9 @@ type Supervisor struct { | |||||
| 434 | cfg SupervisorConfig | 434 | cfg SupervisorConfig | ||
| 435 | logger *slog.Logger | 435 | logger *slog.Logger | ||
| 436 | 436 | ||||
| 437 | - | // pendingPickerKey retains a commit observed while connecting. It is | 437 | + | // picker groups the picker's per-run state: the pending commit key, the |
| 438 | - | // revalidated against the adopted service just like a ready-phase commit. | 438 | + | // preview subscription, and the off-goroutine kill runner. |
| 439 | - | pendingPickerKey string | 439 | + | picker pickerSession |
| 440 | - | preview previewManager | |||
| 441 | 440 | ||||
| 442 | mu sync.Mutex | 441 | mu sync.Mutex | ||
| 443 | state State | 442 | state State | ||
| @@ -468,28 +467,9 @@ type Supervisor struct { | |||||
| 468 | // capabilities is the outer terminal's probed and enabled capabilities, | 467 | // capabilities is the outer terminal's probed and enabled capabilities, | ||
| 469 | // written once in Run before any attachment starts. | 468 | // written once in Run before any attachment starts. | ||
| 470 | capabilities terminalCapabilities | 469 | capabilities terminalCapabilities | ||
| 471 | - | // readySub is the adopted connection's subscription while the ready phase | 470 | + | // nav is the run goroutine's navigation bookkeeping: the ready-phase |
| 472 | - | // runs, so the picker overlay over a live attachment keeps folding broker | 471 | + | // subscription, pending swap and in-place choices, and the route ledger. |
| 473 | - | // publications. It is only touched from the run goroutine. | 472 | + | nav navigationState |
| 474 | - | readySub ports.BrokerSubscription | |||
| 475 | - | // pendingSwap is the request the picker overlay or a daemon navigation | |||
| 476 | - | // committed to another target while an attachment was live. It is only | |||
| 477 | - | // touched from the run goroutine and consumed by settleAttachment. | |||
| 478 | - | pendingSwap *pickerAttachmentTarget | |||
| 479 | - | // pendingInPlace is the choice the live attachment is switching to in | |||
| 480 | - | // place on its own daemon. A refusal turns it into pendingSwap. Only | |||
| 481 | - | // touched from the run goroutine; cleared when the attachment settles. | |||
| 482 | - | pendingInPlace *pendingInPlace | |||
| 483 | - | // inPlaceSeq numbers in-place choices so a late outcome for an older one | |||
| 484 | - | // is never applied to a newer one. | |||
| 485 | - | inPlaceSeq uint64 | |||
| 486 | - | // kills runs the picker's `x` operations off the run goroutine. | |||
| 487 | - | kills pickerKills | |||
| 488 | - | // routes is the client route ledger published to the serving daemon; | |||
| 489 | - | // routesSent is the attachment that received its latest snapshot. Both are | |||
| 490 | - | // only touched from the run goroutine. | |||
| 491 | - | routes *routeLedger | |||
| 492 | - | routesSent AttachmentToken | |||
| 493 | } | 473 | } | ||
| 494 | 474 | ||||
| 495 | // NewSupervisor validates the required dependencies and returns a supervisor | 475 | // NewSupervisor validates the required dependencies and returns a supervisor | ||
| @@ -537,7 +517,7 @@ func NewSupervisor(cfg SupervisorConfig) (*Supervisor, error) { | |||||
| 537 | state: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected}, | 517 | state: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected}, | ||
| 538 | clientID: newClientID(), | 518 | clientID: newClientID(), | ||
| 539 | navigation: cfg.InitialNavigation, | 519 | navigation: cfg.InitialNavigation, | ||
| 540 | - | preview: previewManager{clock: cfg.Clock}, | 520 | + | picker: pickerSession{preview: previewManager{clock: cfg.Clock}}, |
| 541 | } | 521 | } | ||
| 542 | // The host is the supervisor's existing foreground grant. It owns no raw | 522 | // The host is the supervisor's existing foreground grant. It owns no raw | ||
| 543 | // mode and starts no reader: the supervisor keeps its one terminal input | 523 | // mode and starts no reader: the supervisor keeps its one terminal input | ||
| @@ -609,8 +589,8 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||||
| 609 | ) | 589 | ) | ||
| 610 | retire := func() { | 590 | retire := func() { | ||
| 611 | s.retirePickerKill() | 591 | s.retirePickerKill() | ||
| 612 | - | s.preview.close(s.cfg.Picker) | 592 | + | s.picker.preview.close(s.cfg.Picker) |
| 613 | - | s.readySub = nil | 593 | + | s.nav.readySub = nil |
| 614 | if sub != nil { | 594 | if sub != nil { | ||
| 615 | sub.Close() | 595 | sub.Close() | ||
| 616 | sub = nil | 596 | sub = nil | ||
| @@ -679,7 +659,7 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||||
| 679 | } | 659 | } | ||
| 680 | 660 | ||||
| 681 | service, sub = result.service, result.sub | 661 | service, sub = result.service, result.sub | ||
| 682 | - | s.readySub = sub | 662 | + | s.nav.readySub = sub |
| 683 | if s.cfg.Picker != nil { | 663 | if s.cfg.Picker != nil { | ||
| 684 | // Render the first committed publication immediately, before waiting | 664 | // Render the first committed publication immediately, before waiting | ||
| 685 | // for the next one; the picker never shows a stale empty catalogue | 665 | // for the next one; the picker never shows a stale empty catalogue | ||
| @@ -915,8 +895,8 @@ type readyOutcome struct { | |||||
| 915 | // the supervisor can admit exactly the row the user committed. It opens no | 895 | // the supervisor can admit exactly the row the user committed. It opens no | ||
| 916 | // stream and starts no terminal work itself. | 896 | // stream and starts no terminal work itself. | ||
| 917 | func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, sub ports.BrokerSubscription) readyOutcome { | 897 | func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, sub ports.BrokerSubscription) readyOutcome { | ||
| 918 | - | if key := s.pendingPickerKey; key != "" { | 898 | + | if key := s.picker.pendingPickerKey; key != "" { |
| 919 | - | s.pendingPickerKey = "" | 899 | + | s.picker.pendingPickerKey = "" |
| 920 | return readyOutcome{commitKey: key} | 900 | return readyOutcome{commitKey: key} | ||
| 921 | } | 901 | } | ||
| 922 | var changed <-chan struct{} | 902 | var changed <-chan struct{} | ||
| @@ -939,11 +919,11 @@ func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetim | |||||
| 939 | s.cfg.Picker.ApplySnapshot(service.Snapshot()) | 919 | s.cfg.Picker.ApplySnapshot(service.Snapshot()) | ||
| 940 | s.refreshPreview(service) | 920 | s.refreshPreview(service) | ||
| 941 | s.renderCurrent() | 921 | s.renderCurrent() | ||
| 942 | - | case <-s.preview.changed(): | 922 | + | case <-s.picker.preview.changed(): |
| 943 | - | if s.preview.publish(s.cfg.Picker) { | 923 | + | if s.picker.preview.publish(s.cfg.Picker) { |
| 944 | s.renderCurrent() | 924 | s.renderCurrent() | ||
| 945 | } | 925 | } | ||
| 946 | - | case outcome := <-s.kills.results(): | 926 | + | case outcome := <-s.picker.kills.results(): |
| 947 | s.finishPickerKill(service, outcome) | 927 | s.finishPickerKill(service, outcome) | ||
| 948 | case <-ops: | 928 | case <-ops: | ||
| 949 | op, key := s.cfg.Picker.TakeOp() | 929 | op, key := s.cfg.Picker.TakeOp() | ||
| @@ -961,7 +941,7 @@ func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetim | |||||
| 961 | if op.exit { | 941 | if op.exit { | ||
| 962 | // Ctrl+C is the explicit exit. With no attachment to | 942 | // Ctrl+C is the explicit exit. With no attachment to | ||
| 963 | // return to, it is the only close that ends the process. | 943 | // return to, it is the only close that ends the process. | ||
| 964 | - | s.preview.close(s.cfg.Picker) | 944 | + | s.picker.preview.close(s.cfg.Picker) |
| 965 | return readyOutcome{terminated: true} | 945 | return readyOutcome{terminated: true} | ||
| 966 | } | 946 | } | ||
| 967 | // Escape and q cancel; with no attachment there is nothing to | 947 | // Escape and q cancel; with no attachment there is nothing to | ||
| @@ -971,7 +951,7 @@ func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetim | |||||
| 971 | continue | 951 | continue | ||
| 972 | } | 952 | } | ||
| 973 | if op.commit && key != "" { | 953 | if op.commit && key != "" { | ||
| 974 | - | s.preview.close(s.cfg.Picker) | 954 | + | s.picker.preview.close(s.cfg.Picker) |
| 975 | return readyOutcome{commitKey: key} | 955 | return readyOutcome{commitKey: key} | ||
| 976 | } | 956 | } | ||
| 977 | case <-ctx.Done(): | 957 | case <-ctx.Done(): | ||
| @@ -991,10 +971,10 @@ func (s *Supervisor) refreshPreview(service ports.BrokerNavigator) { | |||||
| 991 | } | 971 | } | ||
| 992 | geometry, err := s.cfg.Terminal.Geometry() | 972 | geometry, err := s.cfg.Terminal.Geometry() | ||
| 993 | if err != nil || !geometry.Valid() { | 973 | if err != nil || !geometry.Valid() { | ||
| 994 | - | s.preview.close(s.cfg.Picker) | 974 | + | s.picker.preview.close(s.cfg.Picker) |
| 995 | return | 975 | return | ||
| 996 | } | 976 | } | ||
| 997 | - | s.preview.refresh(service, s.cfg.Picker, geometry.Size) | 977 | + | s.picker.preview.refresh(service, s.cfg.Picker, geometry.Size) |
| 998 | } | 978 | } | ||
| 999 | 979 | ||||
| 1000 | // pickerOps is nil when no picker owns presentation decisions. | 980 | // pickerOps is nil when no picker owns presentation decisions. | ||
| @@ -1011,7 +991,7 @@ func (s *Supervisor) pickerOps() <-chan struct{} { | |||||
| 1011 | func (s *Supervisor) takePickerClose() bool { | 991 | func (s *Supervisor) takePickerClose() bool { | ||
| 1012 | op, key := s.cfg.Picker.TakeOp() | 992 | op, key := s.cfg.Picker.TakeOp() | ||
| 1013 | if op.commit && key != "" { | 993 | if op.commit && key != "" { | ||
| 1014 | - | s.pendingPickerKey = key | 994 | + | s.picker.pendingPickerKey = key |
| 1015 | } | 995 | } | ||
| 1016 | if op.close && !op.exit { | 996 | if op.close && !op.exit { | ||
| 1017 | s.notifyPicker(pickerExitHint) | 997 | s.notifyPicker(pickerExitHint) | ||
M internal/usecase/client/supervisor_attach.go +6 -7
| @@ -477,7 +477,7 @@ func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, inpu | |||
| 477 | 477 | defer timer.Stop() | |
| 478 | 478 | for { | |
| 479 | 479 | select { | |
| 480 | - | case <-s.brokerChanged(): | |
| 480 | + | case <-s.nav.brokerChanged(): | |
| 481 | 481 | snapshot := service.Snapshot() | |
| 482 | 482 | navigation, consumed, err := s.takeInitialNavigation(snapshot) | |
| 483 | 483 | if errors.Is(err, ErrInitialNavigationNotObserved) { | |
| @@ -551,7 +551,7 @@ func (s *Supervisor) runInitialNavigation(ctx context.Context, input *terminalIn | |||
| 551 | 551 | if s.cfg.Picker != nil { | |
| 552 | 552 | s.cfg.Picker.TakeOp() | |
| 553 | 553 | } | |
| 554 | - | s.pendingPickerKey = "" | |
| 554 | + | s.picker.pendingPickerKey = "" | |
| 555 | 555 | return s.runResolvedAttachment(ctx, input, service, pickerAttachmentTarget{request: request}, SessionEnvironmentLocalCLI) | |
| 556 | 556 | } | |
| 557 | 557 | ||
| @@ -878,7 +878,7 @@ settlement: | |||
| 878 | 878 | break settlement | |
| 879 | 879 | case <-s.attachments.NavigationRequests(): | |
| 880 | 880 | overlay.enter() | |
| 881 | - | case <-s.brokerChanged(): | |
| 881 | + | case <-s.nav.brokerChanged(): | |
| 882 | 882 | if overlay.active { | |
| 883 | 883 | overlay.applyPublication() | |
| 884 | 884 | } | |
| @@ -901,7 +901,7 @@ settlement: | |||
| 901 | 901 | overlay.resize() | |
| 902 | 902 | case <-notices.arm(s.cfg.Picker): | |
| 903 | 903 | s.renderCurrent() | |
| 904 | - | case outcome := <-s.kills.results(): | |
| 904 | + | case outcome := <-s.picker.kills.results(): | |
| 905 | 905 | s.finishPickerKill(service, outcome) | |
| 906 | 906 | case <-service.Done(): | |
| 907 | 907 | brokerLost = true | |
| @@ -939,11 +939,10 @@ settlement: | |||
| 939 | 939 | } | |
| 940 | 940 | // The overlay or a daemon navigation may have committed another target | |
| 941 | 941 | // while this attachment was live; this attempt consumes it either way. | |
| 942 | - | swap := s.pendingSwap | |
| 943 | - | s.pendingSwap = nil | |
| 942 | + | swap := s.nav.takeSwap() | |
| 944 | 943 | // An in-place switch still in flight ended with its attachment; the | |
| 945 | 944 | // resume or picker path below decides what comes next. | |
| 946 | - | s.pendingInPlace = nil | |
| 945 | + | s.nav.clearInPlace() | |
| 947 | 946 | if terminated { | |
| 948 | 947 | return attachmentTerminatedOutcome(termErr) | |
| 949 | 948 | } | |
| @@ -477,7 +477,7 @@ func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, inpu | |||||
| 477 | defer timer.Stop() | 477 | defer timer.Stop() | ||
| 478 | for { | 478 | for { | ||
| 479 | select { | 479 | select { | ||
| 480 | - | case <-s.brokerChanged(): | 480 | + | case <-s.nav.brokerChanged(): |
| 481 | snapshot := service.Snapshot() | 481 | snapshot := service.Snapshot() | ||
| 482 | navigation, consumed, err := s.takeInitialNavigation(snapshot) | 482 | navigation, consumed, err := s.takeInitialNavigation(snapshot) | ||
| 483 | if errors.Is(err, ErrInitialNavigationNotObserved) { | 483 | if errors.Is(err, ErrInitialNavigationNotObserved) { | ||
| @@ -551,7 +551,7 @@ func (s *Supervisor) runInitialNavigation(ctx context.Context, input *terminalIn | |||||
| 551 | if s.cfg.Picker != nil { | 551 | if s.cfg.Picker != nil { | ||
| 552 | s.cfg.Picker.TakeOp() | 552 | s.cfg.Picker.TakeOp() | ||
| 553 | } | 553 | } | ||
| 554 | - | s.pendingPickerKey = "" | 554 | + | s.picker.pendingPickerKey = "" |
| 555 | return s.runResolvedAttachment(ctx, input, service, pickerAttachmentTarget{request: request}, SessionEnvironmentLocalCLI) | 555 | return s.runResolvedAttachment(ctx, input, service, pickerAttachmentTarget{request: request}, SessionEnvironmentLocalCLI) | ||
| 556 | } | 556 | } | ||
| 557 | 557 | ||||
| @@ -878,7 +878,7 @@ settlement: | |||||
| 878 | break settlement | 878 | break settlement | ||
| 879 | case <-s.attachments.NavigationRequests(): | 879 | case <-s.attachments.NavigationRequests(): | ||
| 880 | overlay.enter() | 880 | overlay.enter() | ||
| 881 | - | case <-s.brokerChanged(): | 881 | + | case <-s.nav.brokerChanged(): |
| 882 | if overlay.active { | 882 | if overlay.active { | ||
| 883 | overlay.applyPublication() | 883 | overlay.applyPublication() | ||
| 884 | } | 884 | } | ||
| @@ -901,7 +901,7 @@ settlement: | |||||
| 901 | overlay.resize() | 901 | overlay.resize() | ||
| 902 | case <-notices.arm(s.cfg.Picker): | 902 | case <-notices.arm(s.cfg.Picker): | ||
| 903 | s.renderCurrent() | 903 | s.renderCurrent() | ||
| 904 | - | case outcome := <-s.kills.results(): | 904 | + | case outcome := <-s.picker.kills.results(): |
| 905 | s.finishPickerKill(service, outcome) | 905 | s.finishPickerKill(service, outcome) | ||
| 906 | case <-service.Done(): | 906 | case <-service.Done(): | ||
| 907 | brokerLost = true | 907 | brokerLost = true | ||
| @@ -939,11 +939,10 @@ settlement: | |||||
| 939 | } | 939 | } | ||
| 940 | // The overlay or a daemon navigation may have committed another target | 940 | // The overlay or a daemon navigation may have committed another target | ||
| 941 | // while this attachment was live; this attempt consumes it either way. | 941 | // while this attachment was live; this attempt consumes it either way. | ||
| 942 | - | swap := s.pendingSwap | 942 | + | swap := s.nav.takeSwap() |
| 943 | - | s.pendingSwap = nil | |||
| 944 | // An in-place switch still in flight ended with its attachment; the | 943 | // An in-place switch still in flight ended with its attachment; the | ||
| 945 | // resume or picker path below decides what comes next. | 944 | // resume or picker path below decides what comes next. | ||
| 946 | - | s.pendingInPlace = nil | 945 | + | s.nav.clearInPlace() |
| 947 | if terminated { | 946 | if terminated { | ||
| 948 | return attachmentTerminatedOutcome(termErr) | 947 | return attachmentTerminatedOutcome(termErr) | ||
| 949 | } | 948 | } | ||
A internal/usecase/client/supervisor_navigation.go +98 -0
| @@ -0,0 +1,98 @@ | |||
| 1 | + | package client | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "github.com/bnema/vev/internal/ports" | |
| 5 | + | "github.com/bnema/vev/internal/protocol" | |
| 6 | + | ) | |
| 7 | + | ||
| 8 | + | // navigationState is the supervisor's run-goroutine navigation bookkeeping: | |
| 9 | + | // the adopted connection's publication wake, the swap and in-place choices | |
| 10 | + | // still waiting on a live attachment, and the client route ledger. It is only | |
| 11 | + | // touched from the run goroutine, so it has no mutex. The zero value is ready. | |
| 12 | + | type navigationState struct { | |
| 13 | + | // readySub is the adopted connection's subscription while the ready phase | |
| 14 | + | // runs, so the picker overlay over a live attachment keeps folding broker | |
| 15 | + | // publications. | |
| 16 | + | readySub ports.BrokerSubscription | |
| 17 | + | // pendingSwap is the request the picker overlay or a daemon navigation | |
| 18 | + | // committed to another target while an attachment was live. Consumed by | |
| 19 | + | // settleAttachment. | |
| 20 | + | pendingSwap *pickerAttachmentTarget | |
| 21 | + | // pendingInPlace is the choice the live attachment is switching to in | |
| 22 | + | // place on its own daemon. A refusal turns it into pendingSwap. Cleared | |
| 23 | + | // when the attachment settles. | |
| 24 | + | pendingInPlace *pendingInPlace | |
| 25 | + | // inPlaceSeq numbers in-place choices so a late outcome for an older one | |
| 26 | + | // is never applied to a newer one. | |
| 27 | + | inPlaceSeq uint64 | |
| 28 | + | // routes is the client route ledger published to the serving daemon; | |
| 29 | + | // routesSent is the attachment that received its latest snapshot. | |
| 30 | + | routes *routeLedger | |
| 31 | + | routesSent AttachmentToken | |
| 32 | + | } | |
| 33 | + | ||
| 34 | + | // brokerChanged is the adopted connection's publication wake, or nil. | |
| 35 | + | func (n *navigationState) brokerChanged() <-chan struct{} { | |
| 36 | + | if supervisorNil(n.readySub) { | |
| 37 | + | return nil | |
| 38 | + | } | |
| 39 | + | return n.readySub.Changed() | |
| 40 | + | } | |
| 41 | + | ||
| 42 | + | // takeSwap returns and clears the committed swap target. | |
| 43 | + | func (n *navigationState) takeSwap() *pickerAttachmentTarget { | |
| 44 | + | swap := n.pendingSwap | |
| 45 | + | n.pendingSwap = nil | |
| 46 | + | return swap | |
| 47 | + | } | |
| 48 | + | ||
| 49 | + | // switchingInPlaceTo reports whether an in-place switch to target is in flight. | |
| 50 | + | func (n *navigationState) switchingInPlaceTo(target protocol.ExactSessionTarget) bool { | |
| 51 | + | return n.pendingInPlace != nil && n.pendingInPlace.target.request.Target == target | |
| 52 | + | } | |
| 53 | + | ||
| 54 | + | // inPlaceIdle reports whether no in-place switch is in flight. | |
| 55 | + | func (n *navigationState) inPlaceIdle() bool { return n.pendingInPlace == nil } | |
| 56 | + | ||
| 57 | + | // nextInPlaceSeq numbers one new in-place choice. | |
| 58 | + | func (n *navigationState) nextInPlaceSeq() uint64 { | |
| 59 | + | n.inPlaceSeq++ | |
| 60 | + | return n.inPlaceSeq | |
| 61 | + | } | |
| 62 | + | ||
| 63 | + | // startInPlace records the in-place switch numbered seq as in flight. | |
| 64 | + | func (n *navigationState) startInPlace(seq uint64, target pickerAttachmentTarget) { | |
| 65 | + | n.pendingInPlace = &pendingInPlace{seq: seq, target: target} | |
| 66 | + | } | |
| 67 | + | ||
| 68 | + | // clearInPlace forgets any in-place switch in flight. | |
| 69 | + | func (n *navigationState) clearInPlace() { n.pendingInPlace = nil } | |
| 70 | + | ||
| 71 | + | // takeInPlace returns and clears the in-flight in-place switch when it is the | |
| 72 | + | // one numbered seq; an outcome for an older choice returns nil and changes | |
| 73 | + | // nothing. | |
| 74 | + | func (n *navigationState) takeInPlace(seq uint64) *pendingInPlace { | |
| 75 | + | pending := n.pendingInPlace | |
| 76 | + | if pending == nil || pending.seq != seq { | |
| 77 | + | return nil | |
| 78 | + | } | |
| 79 | + | n.pendingInPlace = nil | |
| 80 | + | return pending | |
| 81 | + | } | |
| 82 | + | ||
| 83 | + | // buildRoutes rebuilds the route ledger over the broker snapshot, creating it | |
| 84 | + | // on first use. | |
| 85 | + | func (n *navigationState) buildRoutes(snapshot ports.BrokerSnapshot, active routeActive) (protocol.RecentRouteSnapshot, bool) { | |
| 86 | + | if n.routes == nil { | |
| 87 | + | n.routes = newRouteLedger() | |
| 88 | + | } | |
| 89 | + | return n.routes.build(snapshot, active) | |
| 90 | + | } | |
| 91 | + | ||
| 92 | + | // resolveRoute resolves one route reference against the ledger. | |
| 93 | + | func (n *navigationState) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | |
| 94 | + | if n.routes == nil || ref.IsZero() { | |
| 95 | + | return routeLedgerTarget{}, false | |
| 96 | + | } | |
| 97 | + | return n.routes.resolve(ref) | |
| 98 | + | } | |
| @@ -0,0 +1,98 @@ | |||||
| 1 | + | package client | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "github.com/bnema/vev/internal/ports" | |||
| 5 | + | "github.com/bnema/vev/internal/protocol" | |||
| 6 | + | ) | |||
| 7 | + | ||||
| 8 | + | // navigationState is the supervisor's run-goroutine navigation bookkeeping: | |||
| 9 | + | // the adopted connection's publication wake, the swap and in-place choices | |||
| 10 | + | // still waiting on a live attachment, and the client route ledger. It is only | |||
| 11 | + | // touched from the run goroutine, so it has no mutex. The zero value is ready. | |||
| 12 | + | type navigationState struct { | |||
| 13 | + | // readySub is the adopted connection's subscription while the ready phase | |||
| 14 | + | // runs, so the picker overlay over a live attachment keeps folding broker | |||
| 15 | + | // publications. | |||
| 16 | + | readySub ports.BrokerSubscription | |||
| 17 | + | // pendingSwap is the request the picker overlay or a daemon navigation | |||
| 18 | + | // committed to another target while an attachment was live. Consumed by | |||
| 19 | + | // settleAttachment. | |||
| 20 | + | pendingSwap *pickerAttachmentTarget | |||
| 21 | + | // pendingInPlace is the choice the live attachment is switching to in | |||
| 22 | + | // place on its own daemon. A refusal turns it into pendingSwap. Cleared | |||
| 23 | + | // when the attachment settles. | |||
| 24 | + | pendingInPlace *pendingInPlace | |||
| 25 | + | // inPlaceSeq numbers in-place choices so a late outcome for an older one | |||
| 26 | + | // is never applied to a newer one. | |||
| 27 | + | inPlaceSeq uint64 | |||
| 28 | + | // routes is the client route ledger published to the serving daemon; | |||
| 29 | + | // routesSent is the attachment that received its latest snapshot. | |||
| 30 | + | routes *routeLedger | |||
| 31 | + | routesSent AttachmentToken | |||
| 32 | + | } | |||
| 33 | + | ||||
| 34 | + | // brokerChanged is the adopted connection's publication wake, or nil. | |||
| 35 | + | func (n *navigationState) brokerChanged() <-chan struct{} { | |||
| 36 | + | if supervisorNil(n.readySub) { | |||
| 37 | + | return nil | |||
| 38 | + | } | |||
| 39 | + | return n.readySub.Changed() | |||
| 40 | + | } | |||
| 41 | + | ||||
| 42 | + | // takeSwap returns and clears the committed swap target. | |||
| 43 | + | func (n *navigationState) takeSwap() *pickerAttachmentTarget { | |||
| 44 | + | swap := n.pendingSwap | |||
| 45 | + | n.pendingSwap = nil | |||
| 46 | + | return swap | |||
| 47 | + | } | |||
| 48 | + | ||||
| 49 | + | // switchingInPlaceTo reports whether an in-place switch to target is in flight. | |||
| 50 | + | func (n *navigationState) switchingInPlaceTo(target protocol.ExactSessionTarget) bool { | |||
| 51 | + | return n.pendingInPlace != nil && n.pendingInPlace.target.request.Target == target | |||
| 52 | + | } | |||
| 53 | + | ||||
| 54 | + | // inPlaceIdle reports whether no in-place switch is in flight. | |||
| 55 | + | func (n *navigationState) inPlaceIdle() bool { return n.pendingInPlace == nil } | |||
| 56 | + | ||||
| 57 | + | // nextInPlaceSeq numbers one new in-place choice. | |||
| 58 | + | func (n *navigationState) nextInPlaceSeq() uint64 { | |||
| 59 | + | n.inPlaceSeq++ | |||
| 60 | + | return n.inPlaceSeq | |||
| 61 | + | } | |||
| 62 | + | ||||
| 63 | + | // startInPlace records the in-place switch numbered seq as in flight. | |||
| 64 | + | func (n *navigationState) startInPlace(seq uint64, target pickerAttachmentTarget) { | |||
| 65 | + | n.pendingInPlace = &pendingInPlace{seq: seq, target: target} | |||
| 66 | + | } | |||
| 67 | + | ||||
| 68 | + | // clearInPlace forgets any in-place switch in flight. | |||
| 69 | + | func (n *navigationState) clearInPlace() { n.pendingInPlace = nil } | |||
| 70 | + | ||||
| 71 | + | // takeInPlace returns and clears the in-flight in-place switch when it is the | |||
| 72 | + | // one numbered seq; an outcome for an older choice returns nil and changes | |||
| 73 | + | // nothing. | |||
| 74 | + | func (n *navigationState) takeInPlace(seq uint64) *pendingInPlace { | |||
| 75 | + | pending := n.pendingInPlace | |||
| 76 | + | if pending == nil || pending.seq != seq { | |||
| 77 | + | return nil | |||
| 78 | + | } | |||
| 79 | + | n.pendingInPlace = nil | |||
| 80 | + | return pending | |||
| 81 | + | } | |||
| 82 | + | ||||
| 83 | + | // buildRoutes rebuilds the route ledger over the broker snapshot, creating it | |||
| 84 | + | // on first use. | |||
| 85 | + | func (n *navigationState) buildRoutes(snapshot ports.BrokerSnapshot, active routeActive) (protocol.RecentRouteSnapshot, bool) { | |||
| 86 | + | if n.routes == nil { | |||
| 87 | + | n.routes = newRouteLedger() | |||
| 88 | + | } | |||
| 89 | + | return n.routes.build(snapshot, active) | |||
| 90 | + | } | |||
| 91 | + | ||||
| 92 | + | // resolveRoute resolves one route reference against the ledger. | |||
| 93 | + | func (n *navigationState) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | |||
| 94 | + | if n.routes == nil || ref.IsZero() { | |||
| 95 | + | return routeLedgerTarget{}, false | |||
| 96 | + | } | |||
| 97 | + | return n.routes.resolve(ref) | |||
| 98 | + | } | |||
A internal/usecase/client/supervisor_navigation_test.go +66 -0
| @@ -0,0 +1,66 @@ | |||
| 1 | + | package client | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "testing" | |
| 5 | + | ||
| 6 | + | "github.com/bnema/vev/internal/ports" | |
| 7 | + | "github.com/bnema/vev/internal/protocol" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | func TestNavigationState(t *testing.T) { | |
| 11 | + | tests := []struct { | |
| 12 | + | name string | |
| 13 | + | run func(t *testing.T) | |
| 14 | + | }{ | |
| 15 | + | {"takeInPlace stale seq keeps pending", func(t *testing.T) { | |
| 16 | + | var n navigationState | |
| 17 | + | n.startInPlace(2, pickerAttachmentTarget{}) | |
| 18 | + | if got := n.takeInPlace(1); got != nil { | |
| 19 | + | t.Fatalf("stale takeInPlace = %v, want nil", got) | |
| 20 | + | } | |
| 21 | + | if n.inPlaceIdle() { | |
| 22 | + | t.Fatal("stale takeInPlace cleared pending") | |
| 23 | + | } | |
| 24 | + | }}, | |
| 25 | + | {"takeInPlace matching seq clears", func(t *testing.T) { | |
| 26 | + | var n navigationState | |
| 27 | + | n.startInPlace(2, pickerAttachmentTarget{}) | |
| 28 | + | if got := n.takeInPlace(2); got == nil || got.seq != 2 { | |
| 29 | + | t.Fatalf("takeInPlace = %v, want seq 2", got) | |
| 30 | + | } | |
| 31 | + | if !n.inPlaceIdle() { | |
| 32 | + | t.Fatal("matching takeInPlace kept pending") | |
| 33 | + | } | |
| 34 | + | }}, | |
| 35 | + | {"resolveRoute zero value", func(t *testing.T) { | |
| 36 | + | var n navigationState | |
| 37 | + | if _, ok := n.resolveRoute(protocol.RouteRef{Key: 1, Generation: 1}); ok { | |
| 38 | + | t.Fatal("resolveRoute on zero value = true, want false") | |
| 39 | + | } | |
| 40 | + | }}, | |
| 41 | + | {"buildRoutes creates ledger lazily", func(t *testing.T) { | |
| 42 | + | var n navigationState | |
| 43 | + | if n.routes != nil { | |
| 44 | + | t.Fatal("routes non-nil at zero value") | |
| 45 | + | } | |
| 46 | + | n.buildRoutes(ports.BrokerSnapshot{}, routeActive{}) | |
| 47 | + | if n.routes == nil { | |
| 48 | + | t.Fatal("routes nil after buildRoutes") | |
| 49 | + | } | |
| 50 | + | }}, | |
| 51 | + | {"takeSwap returns and clears", func(t *testing.T) { | |
| 52 | + | var n navigationState | |
| 53 | + | want := &pickerAttachmentTarget{} | |
| 54 | + | n.pendingSwap = want | |
| 55 | + | if got := n.takeSwap(); got != want { | |
| 56 | + | t.Fatalf("takeSwap = %p, want %p", got, want) | |
| 57 | + | } | |
| 58 | + | if n.pendingSwap != nil { | |
| 59 | + | t.Fatal("takeSwap kept swap") | |
| 60 | + | } | |
| 61 | + | }}, | |
| 62 | + | } | |
| 63 | + | for _, tt := range tests { | |
| 64 | + | t.Run(tt.name, tt.run) | |
| 65 | + | } | |
| 66 | + | } | |
| @@ -0,0 +1,66 @@ | |||||
| 1 | + | package client | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "testing" | |||
| 5 | + | ||||
| 6 | + | "github.com/bnema/vev/internal/ports" | |||
| 7 | + | "github.com/bnema/vev/internal/protocol" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | func TestNavigationState(t *testing.T) { | |||
| 11 | + | tests := []struct { | |||
| 12 | + | name string | |||
| 13 | + | run func(t *testing.T) | |||
| 14 | + | }{ | |||
| 15 | + | {"takeInPlace stale seq keeps pending", func(t *testing.T) { | |||
| 16 | + | var n navigationState | |||
| 17 | + | n.startInPlace(2, pickerAttachmentTarget{}) | |||
| 18 | + | if got := n.takeInPlace(1); got != nil { | |||
| 19 | + | t.Fatalf("stale takeInPlace = %v, want nil", got) | |||
| 20 | + | } | |||
| 21 | + | if n.inPlaceIdle() { | |||
| 22 | + | t.Fatal("stale takeInPlace cleared pending") | |||
| 23 | + | } | |||
| 24 | + | }}, | |||
| 25 | + | {"takeInPlace matching seq clears", func(t *testing.T) { | |||
| 26 | + | var n navigationState | |||
| 27 | + | n.startInPlace(2, pickerAttachmentTarget{}) | |||
| 28 | + | if got := n.takeInPlace(2); got == nil || got.seq != 2 { | |||
| 29 | + | t.Fatalf("takeInPlace = %v, want seq 2", got) | |||
| 30 | + | } | |||
| 31 | + | if !n.inPlaceIdle() { | |||
| 32 | + | t.Fatal("matching takeInPlace kept pending") | |||
| 33 | + | } | |||
| 34 | + | }}, | |||
| 35 | + | {"resolveRoute zero value", func(t *testing.T) { | |||
| 36 | + | var n navigationState | |||
| 37 | + | if _, ok := n.resolveRoute(protocol.RouteRef{Key: 1, Generation: 1}); ok { | |||
| 38 | + | t.Fatal("resolveRoute on zero value = true, want false") | |||
| 39 | + | } | |||
| 40 | + | }}, | |||
| 41 | + | {"buildRoutes creates ledger lazily", func(t *testing.T) { | |||
| 42 | + | var n navigationState | |||
| 43 | + | if n.routes != nil { | |||
| 44 | + | t.Fatal("routes non-nil at zero value") | |||
| 45 | + | } | |||
| 46 | + | n.buildRoutes(ports.BrokerSnapshot{}, routeActive{}) | |||
| 47 | + | if n.routes == nil { | |||
| 48 | + | t.Fatal("routes nil after buildRoutes") | |||
| 49 | + | } | |||
| 50 | + | }}, | |||
| 51 | + | {"takeSwap returns and clears", func(t *testing.T) { | |||
| 52 | + | var n navigationState | |||
| 53 | + | want := &pickerAttachmentTarget{} | |||
| 54 | + | n.pendingSwap = want | |||
| 55 | + | if got := n.takeSwap(); got != want { | |||
| 56 | + | t.Fatalf("takeSwap = %p, want %p", got, want) | |||
| 57 | + | } | |||
| 58 | + | if n.pendingSwap != nil { | |||
| 59 | + | t.Fatal("takeSwap kept swap") | |||
| 60 | + | } | |||
| 61 | + | }}, | |||
| 62 | + | } | |||
| 63 | + | for _, tt := range tests { | |||
| 64 | + | t.Run(tt.name, tt.run) | |||
| 65 | + | } | |||
| 66 | + | } | |||
M internal/usecase/client/supervisor_overlay.go +13 -14
| @@ -44,7 +44,7 @@ func (o *attachmentPickerOverlay) previewChanged() <-chan struct{} { | |||
| 44 | 44 | if !o.active { | |
| 45 | 45 | return nil | |
| 46 | 46 | } | |
| 47 | - | return o.sup.preview.changed() | |
| 47 | + | return o.sup.picker.preview.changed() | |
| 48 | 48 | } | |
| 49 | 49 | ||
| 50 | 50 | func (o *attachmentPickerOverlay) ops() <-chan struct{} { | |
| @@ -113,7 +113,7 @@ func (o *attachmentPickerOverlay) drop() { | |||
| 113 | 113 | return | |
| 114 | 114 | } | |
| 115 | 115 | o.active = false | |
| 116 | - | o.sup.preview.close(o.picker()) | |
| 116 | + | o.sup.picker.preview.close(o.picker()) | |
| 117 | 117 | o.picker().SetOwnsInput(false) | |
| 118 | 118 | } | |
| 119 | 119 | ||
| @@ -124,7 +124,7 @@ func (o *attachmentPickerOverlay) applyPublication() { | |||
| 124 | 124 | } | |
| 125 | 125 | ||
| 126 | 126 | func (o *attachmentPickerOverlay) publishPreview() { | |
| 127 | - | if o.sup.preview.publish(o.picker()) { | |
| 127 | + | if o.sup.picker.preview.publish(o.picker()) { | |
| 128 | 128 | o.sup.renderCurrent() | |
| 129 | 129 | } | |
| 130 | 130 | } | |
| @@ -154,12 +154,12 @@ func (o *attachmentPickerOverlay) takeOp() { | |||
| 154 | 154 | // While an in-place switch is in flight the attachment is leaving | |
| 155 | 155 | // its committed session, so choosing that session again is a new | |
| 156 | 156 | // navigation, not a close. | |
| 157 | - | if s.pendingInPlace != nil && s.pendingInPlace.target.request.Target == request.Target && tab.stopped == nil { | |
| 157 | + | if s.nav.switchingInPlaceTo(request.Target) && tab.stopped == nil { | |
| 158 | 158 | // The attachment is already switching there. | |
| 159 | 159 | o.release(false) | |
| 160 | 160 | return | |
| 161 | 161 | } | |
| 162 | - | if s.pendingInPlace == nil && o.sameTarget(request) && tab.stopped == nil { | |
| 162 | + | if s.nav.inPlaceIdle() && o.sameTarget(request) && tab.stopped == nil { | |
| 163 | 163 | // Another tab of the attached session switches in place; the | |
| 164 | 164 | // attachment is never reconnected for it. | |
| 165 | 165 | _, current, _ := s.attachments.committedView() | |
| @@ -179,9 +179,9 @@ func (o *attachmentPickerOverlay) takeOp() { | |||
| 179 | 179 | o.release(false) | |
| 180 | 180 | return | |
| 181 | 181 | } | |
| 182 | - | s.pendingSwap = &target | |
| 182 | + | s.nav.pendingSwap = &target | |
| 183 | 183 | o.swapping = true | |
| 184 | - | s.preview.close(o.picker()) | |
| 184 | + | s.picker.preview.close(o.picker()) | |
| 185 | 185 | s.attachments.requestDetach(o.run.token) | |
| 186 | 186 | case op.kill && key != "": | |
| 187 | 187 | s.startPickerKill(o.service, key) | |
| @@ -246,11 +246,11 @@ func (o *attachmentPickerOverlay) tryInPlace(target pickerAttachmentTarget) bool | |||
| 246 | 246 | // A stopped session needs a restore, which only a fresh attach does. | |
| 247 | 247 | return false | |
| 248 | 248 | } | |
| 249 | - | s.inPlaceSeq++ | |
| 250 | - | if !s.attachments.requestInPlace(o.run.token, inPlaceSwitch{seq: s.inPlaceSeq, target: request.Target, tab: target.tab.preferred}) { | |
| 249 | + | seq := s.nav.nextInPlaceSeq() | |
| 250 | + | if !s.attachments.requestInPlace(o.run.token, inPlaceSwitch{seq: seq, target: request.Target, tab: target.tab.preferred}) { | |
| 251 | 251 | return false | |
| 252 | 252 | } | |
| 253 | - | s.pendingInPlace = &pendingInPlace{seq: s.inPlaceSeq, target: target} | |
| 253 | + | s.nav.startInPlace(seq, target) | |
| 254 | 254 | return true | |
| 255 | 255 | } | |
| 256 | 256 | ||
| @@ -271,18 +271,17 @@ func catalogueSessionLive(snapshot ports.BrokerSnapshot, request ports.BrokerOpe | |||
| 271 | 271 | // superseded switch was replaced by a newer navigation and is dropped. | |
| 272 | 272 | func (o *attachmentPickerOverlay) inPlaceSettled(result inPlaceResult) { | |
| 273 | 273 | s := o.sup | |
| 274 | - | pending := s.pendingInPlace | |
| 275 | - | if pending == nil || pending.seq != result.seq { | |
| 274 | + | pending := s.nav.takeInPlace(result.seq) | |
| 275 | + | if pending == nil { | |
| 276 | 276 | return | |
| 277 | 277 | } | |
| 278 | - | s.pendingInPlace = nil | |
| 279 | 278 | if result.outcome != inPlaceRefused || o.swapping { | |
| 280 | 279 | return | |
| 281 | 280 | } | |
| 282 | 281 | if o.active { | |
| 283 | 282 | o.exit() | |
| 284 | 283 | } | |
| 285 | - | s.pendingSwap = &pending.target | |
| 284 | + | s.nav.pendingSwap = &pending.target | |
| 286 | 285 | o.swapping = true | |
| 287 | 286 | s.attachments.requestDetach(o.run.token) | |
| 288 | 287 | } | |
| @@ -44,7 +44,7 @@ func (o *attachmentPickerOverlay) previewChanged() <-chan struct{} { | |||||
| 44 | if !o.active { | 44 | if !o.active { | ||
| 45 | return nil | 45 | return nil | ||
| 46 | } | 46 | } | ||
| 47 | - | return o.sup.preview.changed() | 47 | + | return o.sup.picker.preview.changed() |
| 48 | } | 48 | } | ||
| 49 | 49 | ||||
| 50 | func (o *attachmentPickerOverlay) ops() <-chan struct{} { | 50 | func (o *attachmentPickerOverlay) ops() <-chan struct{} { | ||
| @@ -113,7 +113,7 @@ func (o *attachmentPickerOverlay) drop() { | |||||
| 113 | return | 113 | return | ||
| 114 | } | 114 | } | ||
| 115 | o.active = false | 115 | o.active = false | ||
| 116 | - | o.sup.preview.close(o.picker()) | 116 | + | o.sup.picker.preview.close(o.picker()) |
| 117 | o.picker().SetOwnsInput(false) | 117 | o.picker().SetOwnsInput(false) | ||
| 118 | } | 118 | } | ||
| 119 | 119 | ||||
| @@ -124,7 +124,7 @@ func (o *attachmentPickerOverlay) applyPublication() { | |||||
| 124 | } | 124 | } | ||
| 125 | 125 | ||||
| 126 | func (o *attachmentPickerOverlay) publishPreview() { | 126 | func (o *attachmentPickerOverlay) publishPreview() { | ||
| 127 | - | if o.sup.preview.publish(o.picker()) { | 127 | + | if o.sup.picker.preview.publish(o.picker()) { |
| 128 | o.sup.renderCurrent() | 128 | o.sup.renderCurrent() | ||
| 129 | } | 129 | } | ||
| 130 | } | 130 | } | ||
| @@ -154,12 +154,12 @@ func (o *attachmentPickerOverlay) takeOp() { | |||||
| 154 | // While an in-place switch is in flight the attachment is leaving | 154 | // While an in-place switch is in flight the attachment is leaving | ||
| 155 | // its committed session, so choosing that session again is a new | 155 | // its committed session, so choosing that session again is a new | ||
| 156 | // navigation, not a close. | 156 | // navigation, not a close. | ||
| 157 | - | if s.pendingInPlace != nil && s.pendingInPlace.target.request.Target == request.Target && tab.stopped == nil { | 157 | + | if s.nav.switchingInPlaceTo(request.Target) && tab.stopped == nil { |
| 158 | // The attachment is already switching there. | 158 | // The attachment is already switching there. | ||
| 159 | o.release(false) | 159 | o.release(false) | ||
| 160 | return | 160 | return | ||
| 161 | } | 161 | } | ||
| 162 | - | if s.pendingInPlace == nil && o.sameTarget(request) && tab.stopped == nil { | 162 | + | if s.nav.inPlaceIdle() && o.sameTarget(request) && tab.stopped == nil { |
| 163 | // Another tab of the attached session switches in place; the | 163 | // Another tab of the attached session switches in place; the | ||
| 164 | // attachment is never reconnected for it. | 164 | // attachment is never reconnected for it. | ||
| 165 | _, current, _ := s.attachments.committedView() | 165 | _, current, _ := s.attachments.committedView() | ||
| @@ -179,9 +179,9 @@ func (o *attachmentPickerOverlay) takeOp() { | |||||
| 179 | o.release(false) | 179 | o.release(false) | ||
| 180 | return | 180 | return | ||
| 181 | } | 181 | } | ||
| 182 | - | s.pendingSwap = &target | 182 | + | s.nav.pendingSwap = &target |
| 183 | o.swapping = true | 183 | o.swapping = true | ||
| 184 | - | s.preview.close(o.picker()) | 184 | + | s.picker.preview.close(o.picker()) |
| 185 | s.attachments.requestDetach(o.run.token) | 185 | s.attachments.requestDetach(o.run.token) | ||
| 186 | case op.kill && key != "": | 186 | case op.kill && key != "": | ||
| 187 | s.startPickerKill(o.service, key) | 187 | s.startPickerKill(o.service, key) | ||
| @@ -246,11 +246,11 @@ func (o *attachmentPickerOverlay) tryInPlace(target pickerAttachmentTarget) bool | |||||
| 246 | // A stopped session needs a restore, which only a fresh attach does. | 246 | // A stopped session needs a restore, which only a fresh attach does. | ||
| 247 | return false | 247 | return false | ||
| 248 | } | 248 | } | ||
| 249 | - | s.inPlaceSeq++ | 249 | + | seq := s.nav.nextInPlaceSeq() |
| 250 | - | if !s.attachments.requestInPlace(o.run.token, inPlaceSwitch{seq: s.inPlaceSeq, target: request.Target, tab: target.tab.preferred}) { | 250 | + | if !s.attachments.requestInPlace(o.run.token, inPlaceSwitch{seq: seq, target: request.Target, tab: target.tab.preferred}) { |
| 251 | return false | 251 | return false | ||
| 252 | } | 252 | } | ||
| 253 | - | s.pendingInPlace = &pendingInPlace{seq: s.inPlaceSeq, target: target} | 253 | + | s.nav.startInPlace(seq, target) |
| 254 | return true | 254 | return true | ||
| 255 | } | 255 | } | ||
| 256 | 256 | ||||
| @@ -271,18 +271,17 @@ func catalogueSessionLive(snapshot ports.BrokerSnapshot, request ports.BrokerOpe | |||||
| 271 | // superseded switch was replaced by a newer navigation and is dropped. | 271 | // superseded switch was replaced by a newer navigation and is dropped. | ||
| 272 | func (o *attachmentPickerOverlay) inPlaceSettled(result inPlaceResult) { | 272 | func (o *attachmentPickerOverlay) inPlaceSettled(result inPlaceResult) { | ||
| 273 | s := o.sup | 273 | s := o.sup | ||
| 274 | - | pending := s.pendingInPlace | 274 | + | pending := s.nav.takeInPlace(result.seq) |
| 275 | - | if pending == nil || pending.seq != result.seq { | 275 | + | if pending == nil { |
| 276 | return | 276 | return | ||
| 277 | } | 277 | } | ||
| 278 | - | s.pendingInPlace = nil | |||
| 279 | if result.outcome != inPlaceRefused || o.swapping { | 278 | if result.outcome != inPlaceRefused || o.swapping { | ||
| 280 | return | 279 | return | ||
| 281 | } | 280 | } | ||
| 282 | if o.active { | 281 | if o.active { | ||
| 283 | o.exit() | 282 | o.exit() | ||
| 284 | } | 283 | } | ||
| 285 | - | s.pendingSwap = &pending.target | 284 | + | s.nav.pendingSwap = &pending.target |
| 286 | o.swapping = true | 285 | o.swapping = true | ||
| 287 | s.attachments.requestDetach(o.run.token) | 286 | s.attachments.requestDetach(o.run.token) | ||
| 288 | } | 287 | } | ||
M internal/usecase/client/supervisor_picker_kill.go +16 -16
| @@ -69,7 +69,7 @@ func (s *Supervisor) startPickerKill(service ports.BrokerNavigator, key string) | |||
| 69 | 69 | if !ok || key == "" || supervisorNil(service) { | |
| 70 | 70 | return | |
| 71 | 71 | } | |
| 72 | - | if s.kills.running { | |
| 72 | + | if s.picker.kills.running { | |
| 73 | 73 | s.notifyPicker(pickerKillRefused("a kill is already in progress")) | |
| 74 | 74 | return | |
| 75 | 75 | } | |
| @@ -82,17 +82,17 @@ func (s *Supervisor) startPickerKill(service ports.BrokerNavigator, key string) | |||
| 82 | 82 | s.notifyPicker(pickerKillRefused("couldn't kill " + target.name + ": broker unavailable")) | |
| 83 | 83 | return | |
| 84 | 84 | } | |
| 85 | - | if s.kills.done == nil { | |
| 86 | - | s.kills.done = make(chan pickerKillOutcome, 1) | |
| 85 | + | if s.picker.kills.done == nil { | |
| 86 | + | s.picker.kills.done = make(chan pickerKillOutcome, 1) | |
| 87 | 87 | } | |
| 88 | 88 | killCtx, cancel := context.WithCancel(context.Background()) | |
| 89 | - | s.kills.running = true | |
| 90 | - | s.kills.cancel = cancel | |
| 91 | - | s.kills.wg.Add(1) | |
| 89 | + | s.picker.kills.running = true | |
| 90 | + | s.picker.kills.cancel = cancel | |
| 91 | + | s.picker.kills.wg.Add(1) | |
| 92 | 92 | go func() { | |
| 93 | - | defer s.kills.wg.Done() | |
| 93 | + | defer s.picker.kills.wg.Done() | |
| 94 | 94 | result, err := operations.Kill(killCtx, target.route, target.name) | |
| 95 | - | s.kills.done <- pickerKillOutcome{target: target, result: result, err: err} | |
| 95 | + | s.picker.kills.done <- pickerKillOutcome{target: target, result: result, err: err} | |
| 96 | 96 | }() | |
| 97 | 97 | } | |
| 98 | 98 | ||
| @@ -110,22 +110,22 @@ func (s *Supervisor) finishPickerKill(service ports.BrokerNavigator, outcome pic | |||
| 110 | 110 | // retirePickerKill cancels and joins an in-flight kill before its borrowed | |
| 111 | 111 | // service closes. Its outcome is still reported, never replayed. | |
| 112 | 112 | func (s *Supervisor) retirePickerKill() { | |
| 113 | - | if !s.kills.running { | |
| 113 | + | if !s.picker.kills.running { | |
| 114 | 114 | return | |
| 115 | 115 | } | |
| 116 | - | s.kills.cancel() | |
| 117 | - | s.kills.wg.Wait() | |
| 118 | - | outcome := <-s.kills.done | |
| 116 | + | s.picker.kills.cancel() | |
| 117 | + | s.picker.kills.wg.Wait() | |
| 118 | + | outcome := <-s.picker.kills.done | |
| 119 | 119 | s.settlePickerKill() | |
| 120 | 120 | s.notifyPicker(pickerKillNotice(outcome)) | |
| 121 | 121 | } | |
| 122 | 122 | ||
| 123 | 123 | func (s *Supervisor) settlePickerKill() { | |
| 124 | - | if s.kills.cancel != nil { | |
| 125 | - | s.kills.cancel() | |
| 124 | + | if s.picker.kills.cancel != nil { | |
| 125 | + | s.picker.kills.cancel() | |
| 126 | 126 | } | |
| 127 | - | s.kills.running = false | |
| 128 | - | s.kills.cancel = nil | |
| 127 | + | s.picker.kills.running = false | |
| 128 | + | s.picker.kills.cancel = nil | |
| 129 | 129 | } | |
| 130 | 130 | ||
| 131 | 131 | // pickerKillRefused is a kill that was never sent. | |
| @@ -69,7 +69,7 @@ func (s *Supervisor) startPickerKill(service ports.BrokerNavigator, key string) | |||||
| 69 | if !ok || key == "" || supervisorNil(service) { | 69 | if !ok || key == "" || supervisorNil(service) { | ||
| 70 | return | 70 | return | ||
| 71 | } | 71 | } | ||
| 72 | - | if s.kills.running { | 72 | + | if s.picker.kills.running { |
| 73 | s.notifyPicker(pickerKillRefused("a kill is already in progress")) | 73 | s.notifyPicker(pickerKillRefused("a kill is already in progress")) | ||
| 74 | return | 74 | return | ||
| 75 | } | 75 | } | ||
| @@ -82,17 +82,17 @@ func (s *Supervisor) startPickerKill(service ports.BrokerNavigator, key string) | |||||
| 82 | s.notifyPicker(pickerKillRefused("couldn't kill " + target.name + ": broker unavailable")) | 82 | s.notifyPicker(pickerKillRefused("couldn't kill " + target.name + ": broker unavailable")) | ||
| 83 | return | 83 | return | ||
| 84 | } | 84 | } | ||
| 85 | - | if s.kills.done == nil { | 85 | + | if s.picker.kills.done == nil { |
| 86 | - | s.kills.done = make(chan pickerKillOutcome, 1) | 86 | + | s.picker.kills.done = make(chan pickerKillOutcome, 1) |
| 87 | } | 87 | } | ||
| 88 | killCtx, cancel := context.WithCancel(context.Background()) | 88 | killCtx, cancel := context.WithCancel(context.Background()) | ||
| 89 | - | s.kills.running = true | 89 | + | s.picker.kills.running = true |
| 90 | - | s.kills.cancel = cancel | 90 | + | s.picker.kills.cancel = cancel |
| 91 | - | s.kills.wg.Add(1) | 91 | + | s.picker.kills.wg.Add(1) |
| 92 | go func() { | 92 | go func() { | ||
| 93 | - | defer s.kills.wg.Done() | 93 | + | defer s.picker.kills.wg.Done() |
| 94 | result, err := operations.Kill(killCtx, target.route, target.name) | 94 | result, err := operations.Kill(killCtx, target.route, target.name) | ||
| 95 | - | s.kills.done <- pickerKillOutcome{target: target, result: result, err: err} | 95 | + | s.picker.kills.done <- pickerKillOutcome{target: target, result: result, err: err} |
| 96 | }() | 96 | }() | ||
| 97 | } | 97 | } | ||
| 98 | 98 | ||||
| @@ -110,22 +110,22 @@ func (s *Supervisor) finishPickerKill(service ports.BrokerNavigator, outcome pic | |||||
| 110 | // retirePickerKill cancels and joins an in-flight kill before its borrowed | 110 | // retirePickerKill cancels and joins an in-flight kill before its borrowed | ||
| 111 | // service closes. Its outcome is still reported, never replayed. | 111 | // service closes. Its outcome is still reported, never replayed. | ||
| 112 | func (s *Supervisor) retirePickerKill() { | 112 | func (s *Supervisor) retirePickerKill() { | ||
| 113 | - | if !s.kills.running { | 113 | + | if !s.picker.kills.running { |
| 114 | return | 114 | return | ||
| 115 | } | 115 | } | ||
| 116 | - | s.kills.cancel() | 116 | + | s.picker.kills.cancel() |
| 117 | - | s.kills.wg.Wait() | 117 | + | s.picker.kills.wg.Wait() |
| 118 | - | outcome := <-s.kills.done | 118 | + | outcome := <-s.picker.kills.done |
| 119 | s.settlePickerKill() | 119 | s.settlePickerKill() | ||
| 120 | s.notifyPicker(pickerKillNotice(outcome)) | 120 | s.notifyPicker(pickerKillNotice(outcome)) | ||
| 121 | } | 121 | } | ||
| 122 | 122 | ||||
| 123 | func (s *Supervisor) settlePickerKill() { | 123 | func (s *Supervisor) settlePickerKill() { | ||
| 124 | - | if s.kills.cancel != nil { | 124 | + | if s.picker.kills.cancel != nil { |
| 125 | - | s.kills.cancel() | 125 | + | s.picker.kills.cancel() |
| 126 | } | 126 | } | ||
| 127 | - | s.kills.running = false | 127 | + | s.picker.kills.running = false |
| 128 | - | s.kills.cancel = nil | 128 | + | s.picker.kills.cancel = nil |
| 129 | } | 129 | } | ||
| 130 | 130 | ||||
| 131 | // pickerKillRefused is a kill that was never sent. | 131 | // pickerKillRefused is a kill that was never sent. | ||
A internal/usecase/client/supervisor_picker_session.go +16 -0
| @@ -0,0 +1,16 @@ | |||
| 1 | + | package client | |
| 2 | + | ||
| 3 | + | // pickerSession groups the picker state the supervisor carries across its run: | |
| 4 | + | // the commit observed while connecting, the selected row's preview | |
| 5 | + | // subscription, and the runner for the picker's kill operations. Like the | |
| 6 | + | // fields it owns, it is only touched from the run goroutine, except the kill | |
| 7 | + | // goroutine's kills.wg/kills.done handoff. Construct with preview.clock set; | |
| 8 | + | // other fields are ready at zero. | |
| 9 | + | type pickerSession struct { | |
| 10 | + | // pendingPickerKey retains a commit observed while connecting. It is | |
| 11 | + | // revalidated against the adopted service just like a ready-phase commit. | |
| 12 | + | pendingPickerKey string | |
| 13 | + | preview previewManager | |
| 14 | + | // kills runs the picker's `x` operations off the run goroutine. | |
| 15 | + | kills pickerKills | |
| 16 | + | } | |
| @@ -0,0 +1,16 @@ | |||||
| 1 | + | package client | |||
| 2 | + | ||||
| 3 | + | // pickerSession groups the picker state the supervisor carries across its run: | |||
| 4 | + | // the commit observed while connecting, the selected row's preview | |||
| 5 | + | // subscription, and the runner for the picker's kill operations. Like the | |||
| 6 | + | // fields it owns, it is only touched from the run goroutine, except the kill | |||
| 7 | + | // goroutine's kills.wg/kills.done handoff. Construct with preview.clock set; | |||
| 8 | + | // other fields are ready at zero. | |||
| 9 | + | type pickerSession struct { | |||
| 10 | + | // pendingPickerKey retains a commit observed while connecting. It is | |||
| 11 | + | // revalidated against the adopted service just like a ready-phase commit. | |||
| 12 | + | pendingPickerKey string | |||
| 13 | + | preview previewManager | |||
| 14 | + | // kills runs the picker's `x` operations off the run goroutine. | |||
| 15 | + | kills pickerKills | |||
| 16 | + | } | |||
M internal/usecase/client/supervisor_routes.go +10 -28
| @@ -31,14 +31,6 @@ import ( | |||
| 31 | 31 | // | |
| 32 | 32 | // Every method runs on the supervisor's run goroutine. | |
| 33 | 33 | ||
| 34 | - | // brokerChanged is the adopted connection's publication wake, or nil. | |
| 35 | - | func (s *Supervisor) brokerChanged() <-chan struct{} { | |
| 36 | - | if supervisorNil(s.readySub) { | |
| 37 | - | return nil | |
| 38 | - | } | |
| 39 | - | return s.readySub.Changed() | |
| 40 | - | } | |
| 41 | - | ||
| 42 | 34 | // publishRoutes rebuilds the ledger over the live attachment and hands a | |
| 43 | 35 | // changed snapshot, or the first one of this attachment, to its worker. | |
| 44 | 36 | func (s *Supervisor) publishRoutes(service ports.BrokerNavigator, run *attachmentRun, request ports.BrokerOpenStreamRequest) { | |
| @@ -46,15 +38,12 @@ func (s *Supervisor) publishRoutes(service ports.BrokerNavigator, run *attachmen | |||
| 46 | 38 | if !known { | |
| 47 | 39 | return | |
| 48 | 40 | } | |
| 49 | - | if s.routes == nil { | |
| 50 | - | s.routes = newRouteLedger() | |
| 51 | - | } | |
| 52 | - | snapshot, changed := s.routes.build(service.Snapshot(), routeActive{known: true, authority: requestAuthority(request), target: target}) | |
| 53 | - | if snapshot.Generation == 0 || (!changed && s.routesSent == run.token) { | |
| 41 | + | snapshot, changed := s.nav.buildRoutes(service.Snapshot(), routeActive{known: true, authority: requestAuthority(request), target: target}) | |
| 42 | + | if snapshot.Generation == 0 || (!changed && s.nav.routesSent == run.token) { | |
| 54 | 43 | return | |
| 55 | 44 | } | |
| 56 | 45 | if s.attachments.publishRoutes(run.token, snapshot) { | |
| 57 | - | s.routesSent = run.token | |
| 46 | + | s.nav.routesSent = run.token | |
| 58 | 47 | } | |
| 59 | 48 | } | |
| 60 | 49 | ||
| @@ -75,12 +64,12 @@ func (s *Supervisor) settleDaemonNavigation(service ports.BrokerNavigator, overl | |||
| 75 | 64 | if actionID := navigationCauseActionID(message); actionID != 0 && s.cfg.UI != nil { | |
| 76 | 65 | s.cfg.UI.follow(overlay.run.fg.uiGeneration, actionID) | |
| 77 | 66 | } | |
| 78 | - | if s.pendingInPlace != nil && s.pendingInPlace.target.request.Target == target.request.Target && target.tab.stopped == nil { | |
| 67 | + | if s.nav.switchingInPlaceTo(target.request.Target) && target.tab.stopped == nil { | |
| 79 | 68 | // The attachment is already switching there. | |
| 80 | 69 | overlay.release(false) | |
| 81 | 70 | return | |
| 82 | 71 | } | |
| 83 | - | if s.pendingInPlace == nil && sameAttachmentTarget(overlay.request, s.attachments.committedTargetOrZero(), target.request) && target.tab.stopped == nil { | |
| 72 | + | if s.nav.inPlaceIdle() && sameAttachmentTarget(overlay.request, s.attachments.committedTargetOrZero(), target.request) && target.tab.stopped == nil { | |
| 84 | 73 | _, current, _ := s.attachments.committedView() | |
| 85 | 74 | if target.tab.preferred != "" && target.tab.preferred != current { | |
| 86 | 75 | s.attachments.requestTabSelection(overlay.run.token, target.tab.preferred) | |
| @@ -96,7 +85,7 @@ func (s *Supervisor) settleDaemonNavigation(service ports.BrokerNavigator, overl | |||
| 96 | 85 | if overlay.active { | |
| 97 | 86 | overlay.exit() | |
| 98 | 87 | } | |
| 99 | - | s.pendingSwap = &target | |
| 88 | + | s.nav.pendingSwap = &target | |
| 100 | 89 | overlay.swapping = true | |
| 101 | 90 | s.attachments.requestDetach(overlay.run.token) | |
| 102 | 91 | } | |
| @@ -119,14 +108,14 @@ func navigationCauseActionID(message protocol.ServerMessage) uint64 { | |||
| 119 | 108 | // daemon detached the source right after sending it. | |
| 120 | 109 | func (s *Supervisor) takeSettledNavigation(service ports.BrokerNavigator, token AttachmentToken, request ports.BrokerOpenStreamRequest) { | |
| 121 | 110 | message, ok := s.attachments.takeNavigation() | |
| 122 | - | if !ok || s.pendingSwap != nil { | |
| 111 | + | if !ok || s.nav.pendingSwap != nil { | |
| 123 | 112 | return | |
| 124 | 113 | } | |
| 125 | 114 | if target, ok := s.resolveDaemonNavigation(service, token, request, message); ok { | |
| 126 | 115 | if actionID := navigationCauseActionID(message); actionID != 0 && s.cfg.UI != nil { | |
| 127 | 116 | s.cfg.UI.follow(s.cfg.UI.generation, actionID) | |
| 128 | 117 | } | |
| 129 | - | s.pendingSwap = &target | |
| 118 | + | s.nav.pendingSwap = &target | |
| 130 | 119 | } | |
| 131 | 120 | } | |
| 132 | 121 | ||
| @@ -139,7 +128,7 @@ func (s *Supervisor) resolveDaemonNavigation(service ports.BrokerNavigator, toke | |||
| 139 | 128 | s.attachments.replyNavigation(token, protocol.RouteNavigationFailure{Key: typed.Key, Generation: typed.Generation, Code: code}) | |
| 140 | 129 | return pickerAttachmentTarget{}, false | |
| 141 | 130 | } | |
| 142 | - | routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) | |
| 131 | + | routed, ok := s.nav.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) | |
| 143 | 132 | if !ok || routed.host { | |
| 144 | 133 | return fail(protocol.RouteFailureStaleSelection) | |
| 145 | 134 | } | |
| @@ -156,7 +145,7 @@ func (s *Supervisor) resolveDaemonNavigation(service ports.BrokerNavigator, toke | |||
| 156 | 145 | s.attachments.replyNavigation(token, protocol.SessionCreationFailure{RequestID: typed.RequestID, Code: code}) | |
| 157 | 146 | return pickerAttachmentTarget{}, false | |
| 158 | 147 | } | |
| 159 | - | routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) | |
| 148 | + | routed, ok := s.nav.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) | |
| 160 | 149 | if !ok { | |
| 161 | 150 | return fail(protocol.RouteFailureStaleSelection) | |
| 162 | 151 | } | |
| @@ -202,13 +191,6 @@ func (s *Supervisor) resolveServingHandoff(service ports.BrokerNavigator, reques | |||
| 202 | 191 | return pickerAttachmentTarget{request: next, tab: attachmentTab{preferred: handoff.PreferredTabID}}, true | |
| 203 | 192 | } | |
| 204 | 193 | ||
| 205 | - | func (s *Supervisor) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | |
| 206 | - | if s.routes == nil || ref.IsZero() { | |
| 207 | - | return routeLedgerTarget{}, false | |
| 208 | - | } | |
| 209 | - | return s.routes.resolve(ref) | |
| 210 | - | } | |
| 211 | - | ||
| 212 | 194 | // resolveRouteSelection resolves one route through the same catalogue rules | |
| 213 | 195 | // as a picker commit, against the latest broker publication. | |
| 214 | 196 | func (s *Supervisor) resolveRouteSelection(service ports.BrokerNavigator, selection pickerSelectionRef) (pickerAttachmentTarget, error) { | |
| @@ -31,14 +31,6 @@ import ( | |||||
| 31 | // | 31 | // | ||
| 32 | // Every method runs on the supervisor's run goroutine. | 32 | // Every method runs on the supervisor's run goroutine. | ||
| 33 | 33 | ||||
| 34 | - | // brokerChanged is the adopted connection's publication wake, or nil. | |||
| 35 | - | func (s *Supervisor) brokerChanged() <-chan struct{} { | |||
| 36 | - | if supervisorNil(s.readySub) { | |||
| 37 | - | return nil | |||
| 38 | - | } | |||
| 39 | - | return s.readySub.Changed() | |||
| 40 | - | } | |||
| 41 | - | ||||
| 42 | // publishRoutes rebuilds the ledger over the live attachment and hands a | 34 | // publishRoutes rebuilds the ledger over the live attachment and hands a | ||
| 43 | // changed snapshot, or the first one of this attachment, to its worker. | 35 | // changed snapshot, or the first one of this attachment, to its worker. | ||
| 44 | func (s *Supervisor) publishRoutes(service ports.BrokerNavigator, run *attachmentRun, request ports.BrokerOpenStreamRequest) { | 36 | func (s *Supervisor) publishRoutes(service ports.BrokerNavigator, run *attachmentRun, request ports.BrokerOpenStreamRequest) { | ||
| @@ -46,15 +38,12 @@ func (s *Supervisor) publishRoutes(service ports.BrokerNavigator, run *attachmen | |||||
| 46 | if !known { | 38 | if !known { | ||
| 47 | return | 39 | return | ||
| 48 | } | 40 | } | ||
| 49 | - | if s.routes == nil { | 41 | + | snapshot, changed := s.nav.buildRoutes(service.Snapshot(), routeActive{known: true, authority: requestAuthority(request), target: target}) |
| 50 | - | s.routes = newRouteLedger() | 42 | + | if snapshot.Generation == 0 || (!changed && s.nav.routesSent == run.token) { |
| 51 | - | } | |||
| 52 | - | snapshot, changed := s.routes.build(service.Snapshot(), routeActive{known: true, authority: requestAuthority(request), target: target}) | |||
| 53 | - | if snapshot.Generation == 0 || (!changed && s.routesSent == run.token) { | |||
| 54 | return | 43 | return | ||
| 55 | } | 44 | } | ||
| 56 | if s.attachments.publishRoutes(run.token, snapshot) { | 45 | if s.attachments.publishRoutes(run.token, snapshot) { | ||
| 57 | - | s.routesSent = run.token | 46 | + | s.nav.routesSent = run.token |
| 58 | } | 47 | } | ||
| 59 | } | 48 | } | ||
| 60 | 49 | ||||
| @@ -75,12 +64,12 @@ func (s *Supervisor) settleDaemonNavigation(service ports.BrokerNavigator, overl | |||||
| 75 | if actionID := navigationCauseActionID(message); actionID != 0 && s.cfg.UI != nil { | 64 | if actionID := navigationCauseActionID(message); actionID != 0 && s.cfg.UI != nil { | ||
| 76 | s.cfg.UI.follow(overlay.run.fg.uiGeneration, actionID) | 65 | s.cfg.UI.follow(overlay.run.fg.uiGeneration, actionID) | ||
| 77 | } | 66 | } | ||
| 78 | - | if s.pendingInPlace != nil && s.pendingInPlace.target.request.Target == target.request.Target && target.tab.stopped == nil { | 67 | + | if s.nav.switchingInPlaceTo(target.request.Target) && target.tab.stopped == nil { |
| 79 | // The attachment is already switching there. | 68 | // The attachment is already switching there. | ||
| 80 | overlay.release(false) | 69 | overlay.release(false) | ||
| 81 | return | 70 | return | ||
| 82 | } | 71 | } | ||
| 83 | - | if s.pendingInPlace == nil && sameAttachmentTarget(overlay.request, s.attachments.committedTargetOrZero(), target.request) && target.tab.stopped == nil { | 72 | + | if s.nav.inPlaceIdle() && sameAttachmentTarget(overlay.request, s.attachments.committedTargetOrZero(), target.request) && target.tab.stopped == nil { |
| 84 | _, current, _ := s.attachments.committedView() | 73 | _, current, _ := s.attachments.committedView() | ||
| 85 | if target.tab.preferred != "" && target.tab.preferred != current { | 74 | if target.tab.preferred != "" && target.tab.preferred != current { | ||
| 86 | s.attachments.requestTabSelection(overlay.run.token, target.tab.preferred) | 75 | s.attachments.requestTabSelection(overlay.run.token, target.tab.preferred) | ||
| @@ -96,7 +85,7 @@ func (s *Supervisor) settleDaemonNavigation(service ports.BrokerNavigator, overl | |||||
| 96 | if overlay.active { | 85 | if overlay.active { | ||
| 97 | overlay.exit() | 86 | overlay.exit() | ||
| 98 | } | 87 | } | ||
| 99 | - | s.pendingSwap = &target | 88 | + | s.nav.pendingSwap = &target |
| 100 | overlay.swapping = true | 89 | overlay.swapping = true | ||
| 101 | s.attachments.requestDetach(overlay.run.token) | 90 | s.attachments.requestDetach(overlay.run.token) | ||
| 102 | } | 91 | } | ||
| @@ -119,14 +108,14 @@ func navigationCauseActionID(message protocol.ServerMessage) uint64 { | |||||
| 119 | // daemon detached the source right after sending it. | 108 | // daemon detached the source right after sending it. | ||
| 120 | func (s *Supervisor) takeSettledNavigation(service ports.BrokerNavigator, token AttachmentToken, request ports.BrokerOpenStreamRequest) { | 109 | func (s *Supervisor) takeSettledNavigation(service ports.BrokerNavigator, token AttachmentToken, request ports.BrokerOpenStreamRequest) { | ||
| 121 | message, ok := s.attachments.takeNavigation() | 110 | message, ok := s.attachments.takeNavigation() | ||
| 122 | - | if !ok || s.pendingSwap != nil { | 111 | + | if !ok || s.nav.pendingSwap != nil { |
| 123 | return | 112 | return | ||
| 124 | } | 113 | } | ||
| 125 | if target, ok := s.resolveDaemonNavigation(service, token, request, message); ok { | 114 | if target, ok := s.resolveDaemonNavigation(service, token, request, message); ok { | ||
| 126 | if actionID := navigationCauseActionID(message); actionID != 0 && s.cfg.UI != nil { | 115 | if actionID := navigationCauseActionID(message); actionID != 0 && s.cfg.UI != nil { | ||
| 127 | s.cfg.UI.follow(s.cfg.UI.generation, actionID) | 116 | s.cfg.UI.follow(s.cfg.UI.generation, actionID) | ||
| 128 | } | 117 | } | ||
| 129 | - | s.pendingSwap = &target | 118 | + | s.nav.pendingSwap = &target |
| 130 | } | 119 | } | ||
| 131 | } | 120 | } | ||
| 132 | 121 | ||||
| @@ -139,7 +128,7 @@ func (s *Supervisor) resolveDaemonNavigation(service ports.BrokerNavigator, toke | |||||
| 139 | s.attachments.replyNavigation(token, protocol.RouteNavigationFailure{Key: typed.Key, Generation: typed.Generation, Code: code}) | 128 | s.attachments.replyNavigation(token, protocol.RouteNavigationFailure{Key: typed.Key, Generation: typed.Generation, Code: code}) | ||
| 140 | return pickerAttachmentTarget{}, false | 129 | return pickerAttachmentTarget{}, false | ||
| 141 | } | 130 | } | ||
| 142 | - | routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) | 131 | + | routed, ok := s.nav.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) |
| 143 | if !ok || routed.host { | 132 | if !ok || routed.host { | ||
| 144 | return fail(protocol.RouteFailureStaleSelection) | 133 | return fail(protocol.RouteFailureStaleSelection) | ||
| 145 | } | 134 | } | ||
| @@ -156,7 +145,7 @@ func (s *Supervisor) resolveDaemonNavigation(service ports.BrokerNavigator, toke | |||||
| 156 | s.attachments.replyNavigation(token, protocol.SessionCreationFailure{RequestID: typed.RequestID, Code: code}) | 145 | s.attachments.replyNavigation(token, protocol.SessionCreationFailure{RequestID: typed.RequestID, Code: code}) | ||
| 157 | return pickerAttachmentTarget{}, false | 146 | return pickerAttachmentTarget{}, false | ||
| 158 | } | 147 | } | ||
| 159 | - | routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) | 148 | + | routed, ok := s.nav.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation}) |
| 160 | if !ok { | 149 | if !ok { | ||
| 161 | return fail(protocol.RouteFailureStaleSelection) | 150 | return fail(protocol.RouteFailureStaleSelection) | ||
| 162 | } | 151 | } | ||
| @@ -202,13 +191,6 @@ func (s *Supervisor) resolveServingHandoff(service ports.BrokerNavigator, reques | |||||
| 202 | return pickerAttachmentTarget{request: next, tab: attachmentTab{preferred: handoff.PreferredTabID}}, true | 191 | return pickerAttachmentTarget{request: next, tab: attachmentTab{preferred: handoff.PreferredTabID}}, true | ||
| 203 | } | 192 | } | ||
| 204 | 193 | ||||
| 205 | - | func (s *Supervisor) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | |||
| 206 | - | if s.routes == nil || ref.IsZero() { | |||
| 207 | - | return routeLedgerTarget{}, false | |||
| 208 | - | } | |||
| 209 | - | return s.routes.resolve(ref) | |||
| 210 | - | } | |||
| 211 | - | ||||
| 212 | // resolveRouteSelection resolves one route through the same catalogue rules | 194 | // resolveRouteSelection resolves one route through the same catalogue rules | ||
| 213 | // as a picker commit, against the latest broker publication. | 195 | // as a picker commit, against the latest broker publication. | ||
| 214 | func (s *Supervisor) resolveRouteSelection(service ports.BrokerNavigator, selection pickerSelectionRef) (pickerAttachmentTarget, error) { | 196 | func (s *Supervisor) resolveRouteSelection(service ports.BrokerNavigator, selection pickerSelectionRef) (pickerAttachmentTarget, error) { | ||