| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | |
| 6 | "reasonix/internal/agent" |
| 7 | "reasonix/internal/control" |
| 8 | ) |
| 9 | |
| 10 | // SetSessionLeases hands the server the session-lease keeper that guards its |
| 11 | // active session file. Call it before serving; a nil keeper leaves gating off. |
| 12 | func (s *Server) SetSessionLeases(k *control.SessionLeaseKeeper) error { |
| 13 | s.leases = k |
| 14 | if k != nil { |
| 15 | k.SetControllerOwnershipBinder(func(ctrl *control.Controller, owner *control.SessionLeaseKeeper) { |
| 16 | s.setControllerLeaseOwner(ctrl, owner) |
| 17 | ctrl.SetOnSessionRecovered(s.sessionRecoveryHandler(ctrl, owner)) |
| 18 | ctrl.SetOnSessionTransition(s.sessionTransitionHandler(ctrl, owner)) |
| 19 | }) |
| 20 | } |
| 21 | if ctrl, ok := s.ctl().(*control.Controller); ok { |
| 22 | if k != nil { |
| 23 | return k.BindControllerAuthority(ctrl) |
| 24 | } |
| 25 | } |
| 26 | return nil |
| 27 | } |
| 28 | |
| 29 | func (s *Server) sessionTransitionHandler(ctrl *control.Controller, k *control.SessionLeaseKeeper) func(control.SessionTransitionInfo) error { |
| 30 | if k == nil && s.tagFor(ctrl) == nil { |
| 31 | return nil |
| 32 | } |
| 33 | return func(info control.SessionTransitionInfo) error { |
| 34 | if k != nil { |
| 35 | if err := k.HandleSessionTransition(info); err != nil { |
| 36 | return err |
| 37 | } |
| 38 | } |
| 39 | path := agent.CanonicalSessionPath(info.TargetPath) |
| 40 | info.OnCommit(func() { |
| 41 | if tag := s.tagFor(ctrl); tag != nil { |
| 42 | if ref, bound := ctrl.SessionRef(); bound { |
| 43 | tag.PrimeIdentity(path, ref.SessionID) |
| 44 | } else { |
| 45 | tag.PrimePath(path) |
| 46 | } |
| 47 | } |
| 48 | if s.publishControllerPathIfCurrent(ctrl, path) && branchTransitionNeedsRouteEvent(info.Reason) { |
| 49 | s.announceSessionChanged(path, false) |
| 50 | } |
| 51 | }) |
| 52 | return nil |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | func branchTransitionNeedsRouteEvent(reason string) bool { |
| 57 | switch reason { |
| 58 | case "fork", "branch", "switch": |
| 59 | return true |
| 60 | default: |
| 61 | return false |
| 62 | } |
| 63 | } |
| 64 | |
| 65 | func (s *Server) sessionRecoveryHandler(ctrl *control.Controller, k *control.SessionLeaseKeeper) func(control.SessionRecoveryInfo) error { |
| 66 | if k == nil && s.tagFor(ctrl) == nil { |
| 67 | return nil |
| 68 | } |
| 69 | return func(info control.SessionRecoveryInfo) error { |
| 70 | if k != nil || s.controllerLeaseOwner(ctrl) != nil { |
| 71 | if err := s.handleControllerSessionRecovered(ctrl, k, info); err != nil { |
| 72 | return err |
| 73 | } |
| 74 | } |
| 75 | if err := s.moveDetachedRecovery(ctrl, info.RecoveryPath); err != nil { |
| 76 | return err |
| 77 | } |
| 78 | info.OnCommit(func() { s.publishRecoveredControllerRoute(ctrl, info.RecoveryPath) }) |
| 79 | return nil |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | func (s *Server) publishRecoveredControllerRoute(ctrl *control.Controller, path string) { |
| 84 | if tag := s.tagFor(ctrl); tag != nil { |
| 85 | if ref, bound := ctrl.SessionRef(); bound { |
| 86 | tag.PrimeIdentity(path, ref.SessionID) |
| 87 | } else { |
| 88 | tag.PrimePath(path) |
| 89 | } |
| 90 | } |
| 91 | if s.publishControllerPathIfCurrent(ctrl, path) { |
| 92 | // Recovery changes foreground identity outside the ordinary transition |
| 93 | // hook. Publish the same must-deliver route barrier so a saturated |
| 94 | // all-session subscriber cannot keep routing later frames to the old path. |
| 95 | s.announceSessionChanged(path, false) |
| 96 | } |
| 97 | } |
| 98 | |
| 99 | func (s *Server) setControllerLeaseOwner(ctrl *control.Controller, owner *control.SessionLeaseKeeper) { |
| 100 | if ctrl == nil { |
| 101 | return |
| 102 | } |
| 103 | s.leaseOwnersMu.Lock() |
| 104 | if owner == nil { |
| 105 | delete(s.leaseOwners, ctrl) |
| 106 | } else { |
| 107 | if s.leaseOwners == nil { |
| 108 | s.leaseOwners = map[*control.Controller]*control.SessionLeaseKeeper{} |
| 109 | } |
| 110 | s.leaseOwners[ctrl] = owner |
| 111 | } |
| 112 | s.leaseOwnersMu.Unlock() |
| 113 | } |
| 114 | |
| 115 | func (s *Server) controllerLeaseOwner(ctrl *control.Controller) *control.SessionLeaseKeeper { |
| 116 | if ctrl == nil { |
| 117 | return nil |
| 118 | } |
| 119 | s.leaseOwnersMu.Lock() |
| 120 | defer s.leaseOwnersMu.Unlock() |
| 121 | return s.leaseOwners[ctrl] |
| 122 | } |
| 123 | |
| 124 | // handleControllerSessionRecovered resolves ownership at invocation time. A |
| 125 | // controller may have captured its callback immediately before a busy switch |
| 126 | // transfers it to a detached keeper; the controller/keeper identity check |
| 127 | // prevents that stale callback from mutating the now-reused foreground keeper. |
| 128 | func (s *Server) handleControllerSessionRecovered(ctrl *control.Controller, fallback *control.SessionLeaseKeeper, info control.SessionRecoveryInfo) error { |
| 129 | owner := s.controllerLeaseOwner(ctrl) |
| 130 | if owner == nil { |
| 131 | owner = fallback |
| 132 | } |
| 133 | for range 3 { |
| 134 | if owner == nil { |
| 135 | return nil |
| 136 | } |
| 137 | handled, err := owner.HandleSessionRecoveredFor(ctrl, info) |
| 138 | if handled { |
| 139 | return err |
| 140 | } |
| 141 | next := s.controllerLeaseOwner(ctrl) |
| 142 | if next == nil || next == owner { |
| 143 | return fmt.Errorf("bind recovery session: controller ownership changed during handoff") |
| 144 | } |
| 145 | owner = next |
| 146 | } |
| 147 | return fmt.Errorf("bind recovery session: controller ownership remained unstable") |
| 148 | } |
| 149 | |
| 150 | // publishControllerPathIfCurrent keeps the controller identity check and its |
| 151 | // broadcaster route update in the same publication critical section. A |
| 152 | // recovery/transition callback from a just-demoted controller therefore |
| 153 | // cannot overwrite the newly published foreground route. |
| 154 | func (s *Server) publishControllerPathIfCurrent(ctrl *control.Controller, path string) bool { |
| 155 | s.mu.Lock() |
| 156 | defer s.mu.Unlock() |
| 157 | if s.ctrl != control.SessionAPI(ctrl) { |
| 158 | return false |
| 159 | } |
| 160 | s.bc.SetCurrentSession(path) |
| 161 | return true |
| 162 | } |
| 163 | |
| 164 | // moveDetachedRecovery keeps the registry key aligned when a background |
| 165 | // controller forks to a recovery transcript after an autosave conflict. |
| 166 | func (s *Server) moveDetachedRecovery(ctrl *control.Controller, recoveryPath string) error { |
| 167 | if ctrl == nil { |
| 168 | return nil |
| 169 | } |
| 170 | recoveryPath = agent.CanonicalSessionPath(recoveryPath) |
| 171 | if recoveryPath == "" { |
| 172 | return nil |
| 173 | } |
| 174 | s.detachedMu.Lock() |
| 175 | defer s.detachedMu.Unlock() |
| 176 | for oldPath, detached := range s.detached { |
| 177 | if detached.ctrl != control.SessionAPI(ctrl) { |
| 178 | continue |
| 179 | } |
| 180 | if existing := s.detached[recoveryPath]; existing != nil && existing != detached { |
| 181 | return fmt.Errorf("recovery session is already running in the background") |
| 182 | } |
| 183 | delete(s.detached, oldPath) |
| 184 | detached.path = recoveryPath |
| 185 | s.detached[recoveryPath] = detached |
| 186 | return nil |
| 187 | } |
| 188 | return nil |
| 189 | } |
| 190 | |
| 191 | // rebindSessionLease moves the server's session lease to path and rebinds the |
| 192 | // write authority generation. A nil keeper gates nothing (tests, embedded use). |
| 193 | func (s *Server) rebindSessionLease(path string) error { |
| 194 | ctrl, _ := s.ctl().(*control.Controller) |
| 195 | return s.rebindSessionLeaseFor(path, ctrl) |
| 196 | } |
| 197 | |
| 198 | func (s *Server) rebindSessionLeaseFor(path string, ctrl *control.Controller) error { |
| 199 | if s.leases == nil { |
| 200 | return nil |
| 201 | } |
| 202 | if err := s.leases.Rebind(path); err != nil { |
| 203 | return err |
| 204 | } |
| 205 | if ctrl != nil { |
| 206 | return s.leases.BindControllerAuthority(ctrl) |
| 207 | } |
| 208 | return nil |
| 209 | } |
| 210 |