refactor(client): extract supervisor navigation state
- commit
- 4992fe8242b9f11abf659bf7ab4050fdd2d106d0
- author
- brice · · 2026-10-08 05:57 UTC
- committer
- brice · · 2026-10-08 08:12 UTC
- parent
- 9ff586d
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +144 -60
M internal/usecase/client/supervisor.go +5 -22
| @@ -468,28 +468,11 @@ type Supervisor struct { | |||
| 468 | 468 | // capabilities is the outer terminal's probed and enabled capabilities, | |
| 469 | 469 | // written once in Run before any attachment starts. | |
| 470 | 470 | 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 | |
| 471 | + | // nav is the run goroutine's navigation bookkeeping: the ready-phase | |
| 472 | + | // subscription, pending swap and in-place choices, and the route ledger. | |
| 473 | + | nav navigationState | |
| 486 | 474 | // kills runs the picker's `x` operations off the run goroutine. | |
| 487 | 475 | 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 | 476 | } | |
| 494 | 477 | ||
| 495 | 478 | // NewSupervisor validates the required dependencies and returns a supervisor | |
| @@ -610,7 +593,7 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||
| 610 | 593 | retire := func() { | |
| 611 | 594 | s.retirePickerKill() | |
| 612 | 595 | s.preview.close(s.cfg.Picker) | |
| 613 | - | s.readySub = nil | |
| 596 | + | s.nav.clearReady() | |
| 614 | 597 | if sub != nil { | |
| 615 | 598 | sub.Close() | |
| 616 | 599 | sub = nil | |
| @@ -679,7 +662,7 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||
| 679 | 662 | } | |
| 680 | 663 | ||
| 681 | 664 | service, sub = result.service, result.sub | |
| 682 | - | s.readySub = sub | |
| 665 | + | s.nav.setReady(sub) | |
| 683 | 666 | if s.cfg.Picker != nil { | |
| 684 | 667 | // Render the first committed publication immediately, before waiting | |
| 685 | 668 | // for the next one; the picker never shows a stale empty catalogue | |
| @@ -468,28 +468,11 @@ type Supervisor struct { | |||||
| 468 | // capabilities is the outer terminal's probed and enabled capabilities, | 468 | // capabilities is the outer terminal's probed and enabled capabilities, | ||
| 469 | // written once in Run before any attachment starts. | 469 | // written once in Run before any attachment starts. | ||
| 470 | capabilities terminalCapabilities | 470 | capabilities terminalCapabilities | ||
| 471 | - | // readySub is the adopted connection's subscription while the ready phase | 471 | + | // nav is the run goroutine's navigation bookkeeping: the ready-phase |
| 472 | - | // runs, so the picker overlay over a live attachment keeps folding broker | 472 | + | // subscription, pending swap and in-place choices, and the route ledger. |
| 473 | - | // publications. It is only touched from the run goroutine. | 473 | + | 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. | 474 | // kills runs the picker's `x` operations off the run goroutine. | ||
| 487 | kills pickerKills | 475 | 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 | } | 476 | } | ||
| 494 | 477 | ||||
| 495 | // NewSupervisor validates the required dependencies and returns a supervisor | 478 | // NewSupervisor validates the required dependencies and returns a supervisor | ||
| @@ -610,7 +593,7 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||||
| 610 | retire := func() { | 593 | retire := func() { | ||
| 611 | s.retirePickerKill() | 594 | s.retirePickerKill() | ||
| 612 | s.preview.close(s.cfg.Picker) | 595 | s.preview.close(s.cfg.Picker) | ||
| 613 | - | s.readySub = nil | 596 | + | s.nav.clearReady() |
| 614 | if sub != nil { | 597 | if sub != nil { | ||
| 615 | sub.Close() | 598 | sub.Close() | ||
| 616 | sub = nil | 599 | sub = nil | ||
| @@ -679,7 +662,7 @@ func (s *Supervisor) Run(ctx context.Context) (retErr error) { | |||||
| 679 | } | 662 | } | ||
| 680 | 663 | ||||
| 681 | service, sub = result.service, result.sub | 664 | service, sub = result.service, result.sub | ||
| 682 | - | s.readySub = sub | 665 | + | s.nav.setReady(sub) |
| 683 | if s.cfg.Picker != nil { | 666 | if s.cfg.Picker != nil { | ||
| 684 | // Render the first committed publication immediately, before waiting | 667 | // Render the first committed publication immediately, before waiting | ||
| 685 | // for the next one; the picker never shows a stale empty catalogue | 668 | // for the next one; the picker never shows a stale empty catalogue | ||
M internal/usecase/client/supervisor_attach.go +4 -5
| @@ -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) { | |
| @@ -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 | } | |
| @@ -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) { | ||
| @@ -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 | } | ||
| @@ -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 +117 -0
| @@ -0,0 +1,117 @@ | |||
| 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. It is only touched from the run goroutine. | |
| 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. It is only | |
| 19 | + | // touched from the run goroutine and consumed by 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. Only | |
| 23 | + | // touched from the run goroutine; cleared 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. Both are | |
| 30 | + | // only touched from the run goroutine. | |
| 31 | + | routes *routeLedger | |
| 32 | + | routesSent AttachmentToken | |
| 33 | + | } | |
| 34 | + | ||
| 35 | + | // brokerChanged is the adopted connection's publication wake, or nil. | |
| 36 | + | func (n *navigationState) brokerChanged() <-chan struct{} { | |
| 37 | + | if supervisorNil(n.readySub) { | |
| 38 | + | return nil | |
| 39 | + | } | |
| 40 | + | return n.readySub.Changed() | |
| 41 | + | } | |
| 42 | + | ||
| 43 | + | // setReady records the adopted connection's subscription. | |
| 44 | + | func (n *navigationState) setReady(sub ports.BrokerSubscription) { n.readySub = sub } | |
| 45 | + | ||
| 46 | + | // clearReady forgets the adopted connection's subscription. | |
| 47 | + | func (n *navigationState) clearReady() { n.readySub = nil } | |
| 48 | + | ||
| 49 | + | // setSwap commits the next attachment target. | |
| 50 | + | func (n *navigationState) setSwap(target *pickerAttachmentTarget) { n.pendingSwap = target } | |
| 51 | + | ||
| 52 | + | // hasSwap reports whether a swap target is already committed. | |
| 53 | + | func (n *navigationState) hasSwap() bool { return n.pendingSwap != nil } | |
| 54 | + | ||
| 55 | + | // takeSwap returns and clears the committed swap target. | |
| 56 | + | func (n *navigationState) takeSwap() *pickerAttachmentTarget { | |
| 57 | + | swap := n.pendingSwap | |
| 58 | + | n.pendingSwap = nil | |
| 59 | + | return swap | |
| 60 | + | } | |
| 61 | + | ||
| 62 | + | // switchingInPlaceTo reports whether an in-place switch to target is in flight. | |
| 63 | + | func (n *navigationState) switchingInPlaceTo(target protocol.ExactSessionTarget) bool { | |
| 64 | + | return n.pendingInPlace != nil && n.pendingInPlace.target.request.Target == target | |
| 65 | + | } | |
| 66 | + | ||
| 67 | + | // inPlaceIdle reports whether no in-place switch is in flight. | |
| 68 | + | func (n *navigationState) inPlaceIdle() bool { return n.pendingInPlace == nil } | |
| 69 | + | ||
| 70 | + | // nextInPlaceSeq numbers one new in-place choice. | |
| 71 | + | func (n *navigationState) nextInPlaceSeq() uint64 { | |
| 72 | + | n.inPlaceSeq++ | |
| 73 | + | return n.inPlaceSeq | |
| 74 | + | } | |
| 75 | + | ||
| 76 | + | // startInPlace records the in-place switch numbered seq as in flight. | |
| 77 | + | func (n *navigationState) startInPlace(seq uint64, target pickerAttachmentTarget) { | |
| 78 | + | n.pendingInPlace = &pendingInPlace{seq: seq, target: target} | |
| 79 | + | } | |
| 80 | + | ||
| 81 | + | // clearInPlace forgets any in-place switch in flight. | |
| 82 | + | func (n *navigationState) clearInPlace() { n.pendingInPlace = nil } | |
| 83 | + | ||
| 84 | + | // takeInPlace returns and clears the in-flight in-place switch when it is the | |
| 85 | + | // one numbered seq; an outcome for an older choice returns nil and changes | |
| 86 | + | // nothing. | |
| 87 | + | func (n *navigationState) takeInPlace(seq uint64) *pendingInPlace { | |
| 88 | + | pending := n.pendingInPlace | |
| 89 | + | if pending == nil || pending.seq != seq { | |
| 90 | + | return nil | |
| 91 | + | } | |
| 92 | + | n.pendingInPlace = nil | |
| 93 | + | return pending | |
| 94 | + | } | |
| 95 | + | ||
| 96 | + | // buildRoutes rebuilds the route ledger over the broker snapshot, creating it | |
| 97 | + | // on first use. | |
| 98 | + | func (n *navigationState) buildRoutes(snapshot ports.BrokerSnapshot, active routeActive) (protocol.RecentRouteSnapshot, bool) { | |
| 99 | + | if n.routes == nil { | |
| 100 | + | n.routes = newRouteLedger() | |
| 101 | + | } | |
| 102 | + | return n.routes.build(snapshot, active) | |
| 103 | + | } | |
| 104 | + | ||
| 105 | + | // routesSentTo reports whether token already received a route snapshot. | |
| 106 | + | func (n *navigationState) routesSentTo(token AttachmentToken) bool { return n.routesSent == token } | |
| 107 | + | ||
| 108 | + | // markRoutesSent records the attachment that received the latest snapshot. | |
| 109 | + | func (n *navigationState) markRoutesSent(token AttachmentToken) { n.routesSent = token } | |
| 110 | + | ||
| 111 | + | // resolveRoute resolves one route reference against the ledger. | |
| 112 | + | func (n *navigationState) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | |
| 113 | + | if n.routes == nil || ref.IsZero() { | |
| 114 | + | return routeLedgerTarget{}, false | |
| 115 | + | } | |
| 116 | + | return n.routes.resolve(ref) | |
| 117 | + | } | |
| @@ -0,0 +1,117 @@ | |||||
| 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. It is only touched from the run goroutine. | |||
| 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. It is only | |||
| 19 | + | // touched from the run goroutine and consumed by 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. Only | |||
| 23 | + | // touched from the run goroutine; cleared 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. Both are | |||
| 30 | + | // only touched from the run goroutine. | |||
| 31 | + | routes *routeLedger | |||
| 32 | + | routesSent AttachmentToken | |||
| 33 | + | } | |||
| 34 | + | ||||
| 35 | + | // brokerChanged is the adopted connection's publication wake, or nil. | |||
| 36 | + | func (n *navigationState) brokerChanged() <-chan struct{} { | |||
| 37 | + | if supervisorNil(n.readySub) { | |||
| 38 | + | return nil | |||
| 39 | + | } | |||
| 40 | + | return n.readySub.Changed() | |||
| 41 | + | } | |||
| 42 | + | ||||
| 43 | + | // setReady records the adopted connection's subscription. | |||
| 44 | + | func (n *navigationState) setReady(sub ports.BrokerSubscription) { n.readySub = sub } | |||
| 45 | + | ||||
| 46 | + | // clearReady forgets the adopted connection's subscription. | |||
| 47 | + | func (n *navigationState) clearReady() { n.readySub = nil } | |||
| 48 | + | ||||
| 49 | + | // setSwap commits the next attachment target. | |||
| 50 | + | func (n *navigationState) setSwap(target *pickerAttachmentTarget) { n.pendingSwap = target } | |||
| 51 | + | ||||
| 52 | + | // hasSwap reports whether a swap target is already committed. | |||
| 53 | + | func (n *navigationState) hasSwap() bool { return n.pendingSwap != nil } | |||
| 54 | + | ||||
| 55 | + | // takeSwap returns and clears the committed swap target. | |||
| 56 | + | func (n *navigationState) takeSwap() *pickerAttachmentTarget { | |||
| 57 | + | swap := n.pendingSwap | |||
| 58 | + | n.pendingSwap = nil | |||
| 59 | + | return swap | |||
| 60 | + | } | |||
| 61 | + | ||||
| 62 | + | // switchingInPlaceTo reports whether an in-place switch to target is in flight. | |||
| 63 | + | func (n *navigationState) switchingInPlaceTo(target protocol.ExactSessionTarget) bool { | |||
| 64 | + | return n.pendingInPlace != nil && n.pendingInPlace.target.request.Target == target | |||
| 65 | + | } | |||
| 66 | + | ||||
| 67 | + | // inPlaceIdle reports whether no in-place switch is in flight. | |||
| 68 | + | func (n *navigationState) inPlaceIdle() bool { return n.pendingInPlace == nil } | |||
| 69 | + | ||||
| 70 | + | // nextInPlaceSeq numbers one new in-place choice. | |||
| 71 | + | func (n *navigationState) nextInPlaceSeq() uint64 { | |||
| 72 | + | n.inPlaceSeq++ | |||
| 73 | + | return n.inPlaceSeq | |||
| 74 | + | } | |||
| 75 | + | ||||
| 76 | + | // startInPlace records the in-place switch numbered seq as in flight. | |||
| 77 | + | func (n *navigationState) startInPlace(seq uint64, target pickerAttachmentTarget) { | |||
| 78 | + | n.pendingInPlace = &pendingInPlace{seq: seq, target: target} | |||
| 79 | + | } | |||
| 80 | + | ||||
| 81 | + | // clearInPlace forgets any in-place switch in flight. | |||
| 82 | + | func (n *navigationState) clearInPlace() { n.pendingInPlace = nil } | |||
| 83 | + | ||||
| 84 | + | // takeInPlace returns and clears the in-flight in-place switch when it is the | |||
| 85 | + | // one numbered seq; an outcome for an older choice returns nil and changes | |||
| 86 | + | // nothing. | |||
| 87 | + | func (n *navigationState) takeInPlace(seq uint64) *pendingInPlace { | |||
| 88 | + | pending := n.pendingInPlace | |||
| 89 | + | if pending == nil || pending.seq != seq { | |||
| 90 | + | return nil | |||
| 91 | + | } | |||
| 92 | + | n.pendingInPlace = nil | |||
| 93 | + | return pending | |||
| 94 | + | } | |||
| 95 | + | ||||
| 96 | + | // buildRoutes rebuilds the route ledger over the broker snapshot, creating it | |||
| 97 | + | // on first use. | |||
| 98 | + | func (n *navigationState) buildRoutes(snapshot ports.BrokerSnapshot, active routeActive) (protocol.RecentRouteSnapshot, bool) { | |||
| 99 | + | if n.routes == nil { | |||
| 100 | + | n.routes = newRouteLedger() | |||
| 101 | + | } | |||
| 102 | + | return n.routes.build(snapshot, active) | |||
| 103 | + | } | |||
| 104 | + | ||||
| 105 | + | // routesSentTo reports whether token already received a route snapshot. | |||
| 106 | + | func (n *navigationState) routesSentTo(token AttachmentToken) bool { return n.routesSent == token } | |||
| 107 | + | ||||
| 108 | + | // markRoutesSent records the attachment that received the latest snapshot. | |||
| 109 | + | func (n *navigationState) markRoutesSent(token AttachmentToken) { n.routesSent = token } | |||
| 110 | + | ||||
| 111 | + | // resolveRoute resolves one route reference against the ledger. | |||
| 112 | + | func (n *navigationState) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | |||
| 113 | + | if n.routes == nil || ref.IsZero() { | |||
| 114 | + | return routeLedgerTarget{}, false | |||
| 115 | + | } | |||
| 116 | + | return n.routes.resolve(ref) | |||
| 117 | + | } | |||
M internal/usecase/client/supervisor_overlay.go +9 -10
| @@ -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,7 +179,7 @@ func (o *attachmentPickerOverlay) takeOp() { | |||
| 179 | 179 | o.release(false) | |
| 180 | 180 | return | |
| 181 | 181 | } | |
| 182 | - | s.pendingSwap = &target | |
| 182 | + | s.nav.setSwap(&target) | |
| 183 | 183 | o.swapping = true | |
| 184 | 184 | s.preview.close(o.picker()) | |
| 185 | 185 | s.attachments.requestDetach(o.run.token) | |
| @@ -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.setSwap(&pending.target) | |
| 286 | 285 | o.swapping = true | |
| 287 | 286 | s.attachments.requestDetach(o.run.token) | |
| 288 | 287 | } | |
| @@ -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,7 +179,7 @@ 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.setSwap(&target) |
| 183 | o.swapping = true | 183 | o.swapping = true | ||
| 184 | s.preview.close(o.picker()) | 184 | s.preview.close(o.picker()) | ||
| 185 | s.attachments.requestDetach(o.run.token) | 185 | s.attachments.requestDetach(o.run.token) | ||
| @@ -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.setSwap(&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_routes.go +9 -23
| @@ -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.routesSentTo(run.token)) { | |
| 54 | 43 | return | |
| 55 | 44 | } | |
| 56 | 45 | if s.attachments.publishRoutes(run.token, snapshot) { | |
| 57 | - | s.routesSent = run.token | |
| 46 | + | s.nav.markRoutesSent(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.setSwap(&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.hasSwap() { | |
| 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.setSwap(&target) | |
| 130 | 119 | } | |
| 131 | 120 | } | |
| 132 | 121 | ||
| @@ -203,10 +192,7 @@ func (s *Supervisor) resolveServingHandoff(service ports.BrokerNavigator, reques | |||
| 203 | 192 | } | |
| 204 | 193 | ||
| 205 | 194 | 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) | |
| 195 | + | return s.nav.resolveRoute(ref) | |
| 210 | 196 | } | |
| 211 | 197 | ||
| 212 | 198 | // resolveRouteSelection resolves one route through the same catalogue rules | |
| @@ -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.routesSentTo(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.markRoutesSent(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.setSwap(&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.hasSwap() { |
| 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.setSwap(&target) |
| 130 | } | 119 | } | ||
| 131 | } | 120 | } | ||
| 132 | 121 | ||||
| @@ -203,10 +192,7 @@ func (s *Supervisor) resolveServingHandoff(service ports.BrokerNavigator, reques | |||||
| 203 | } | 192 | } | ||
| 204 | 193 | ||||
| 205 | func (s *Supervisor) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | 194 | func (s *Supervisor) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) { | ||
| 206 | - | if s.routes == nil || ref.IsZero() { | 195 | + | return s.nav.resolveRoute(ref) |
| 207 | - | return routeLedgerTarget{}, false | |||
| 208 | - | } | |||
| 209 | - | return s.routes.resolve(ref) | |||
| 210 | } | 196 | } | ||
| 211 | 197 | ||||
| 212 | // resolveRouteSelection resolves one route through the same catalogue rules | 198 | // resolveRouteSelection resolves one route through the same catalogue rules | ||