| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "log/slog" |
| 8 | "maps" |
| 9 | "net/http" |
| 10 | "reflect" |
| 11 | "sort" |
| 12 | "sync" |
| 13 | "sync/atomic" |
| 14 | "time" |
| 15 | |
| 16 | "reasonix/internal/event" |
| 17 | ) |
| 18 | |
| 19 | type remoteRuntimeSync struct { |
| 20 | mu sync.Mutex |
| 21 | pending bool |
| 22 | unsupported map[string]string |
| 23 | } |
| 24 | |
| 25 | var remoteRuntimeDiagnostics struct { |
| 26 | stale, conflicts, failures atomic.Uint64 |
| 27 | } |
| 28 | |
| 29 | func validRuntimeState(state event.RuntimeStateSnapshot) bool { |
| 30 | return state.SchemaVersion == 1 && state.RuntimeEpoch != "" && state.Revision > 0 && state.BackgroundJobs >= 0 && |
| 31 | state.DurableSeq <= state.CommittedSeq && |
| 32 | (state.Phase == "idle" || state.Phase == "executing" || state.Phase == "finishing" || state.Phase == "cancelling" || state.Phase == "recovery_required" || state.Phase == "closed") |
| 33 | } |
| 34 | |
| 35 | // acceptRemoteRuntimeStateLocked shares source ordering for GET and SSE. Host |
| 36 | // generation and route validation remain the caller's responsibility. |
| 37 | func acceptRemoteRuntimeStateLocked(tab *remoteTab, path string, next event.RuntimeStateSnapshot, authoritative bool) bool { |
| 38 | if !validRuntimeState(next) { |
| 39 | return false |
| 40 | } |
| 41 | if path == "" { |
| 42 | path = tab.routing.currentPath |
| 43 | } |
| 44 | if tab.runtimeStates == nil { |
| 45 | tab.runtimeStates = map[string]event.RuntimeStateSnapshot{} |
| 46 | } |
| 47 | previous, found := tab.runtimeStates[path] |
| 48 | if found && previous.RuntimeEpoch != next.RuntimeEpoch && !authoritative { |
| 49 | return false |
| 50 | } |
| 51 | if !found && !authoritative { |
| 52 | return false |
| 53 | } |
| 54 | if found && previous.RuntimeEpoch == next.RuntimeEpoch { |
| 55 | if next.Revision < previous.Revision { |
| 56 | slog.Debug("remote runtime stale snapshot discarded", "source", "serve", "revision", next.Revision, "stale", remoteRuntimeDiagnostics.stale.Add(1)) |
| 57 | return false |
| 58 | } |
| 59 | if next.Revision == previous.Revision { |
| 60 | if !reflect.DeepEqual(previous, next) { |
| 61 | slog.Warn("remote runtime snapshot version conflict", "source", "serve", "revision", next.Revision, "conflicts", remoteRuntimeDiagnostics.conflicts.Add(1)) |
| 62 | return false |
| 63 | } |
| 64 | if !authoritative || tab.runtimeUnknown[path] == 0 { |
| 65 | return false |
| 66 | } |
| 67 | } |
| 68 | } |
| 69 | tab.runtimeStates[path] = next |
| 70 | tab.runtime.syncFailed = false |
| 71 | delete(tab.runtimeUnknown, path) |
| 72 | delete(tab.runtimeConflicts, path) |
| 73 | if path == tab.routing.currentPath { |
| 74 | tab.runtime.snapshot = next |
| 75 | tab.runtime.running, tab.runtime.pendingPrompt = next.Running, next.PendingPrompt |
| 76 | tab.runtime.backgroundJobs, tab.runtime.cancellable, tab.runtime.cancelRequested = next.BackgroundJobs, next.Cancellable, next.CancelRequested |
| 77 | } |
| 78 | if tab.routing.running == nil { |
| 79 | tab.routing.running = map[string]bool{} |
| 80 | } |
| 81 | tab.routing.running[path] = next.ActiveWork() |
| 82 | tab.runtime.revision++ |
| 83 | return true |
| 84 | } |
| 85 | |
| 86 | func (a *App) acceptRemoteRuntimeFrame(tabID string, gen uint64, path string, frame json.RawMessage) { |
| 87 | var payload struct { |
| 88 | State json.RawMessage `json:"runtimeState"` |
| 89 | } |
| 90 | if json.Unmarshal(frame, &payload) != nil || len(payload.State) == 0 { |
| 91 | return |
| 92 | } |
| 93 | state, decodeErr := decodeRemoteRuntimeState(payload.State) |
| 94 | a.remoteTabMu.Lock() |
| 95 | tab := a.remoteTabs[tabID] |
| 96 | a.remoteTabMu.Unlock() |
| 97 | if tab == nil { |
| 98 | return |
| 99 | } |
| 100 | tab.routeEventMu.Lock() |
| 101 | defer tab.routeEventMu.Unlock() |
| 102 | a.remoteTabMu.Lock() |
| 103 | if a.remoteTabs[tabID] != tab || tab.gen != gen { |
| 104 | a.remoteTabMu.Unlock() |
| 105 | return |
| 106 | } |
| 107 | if path == "" { |
| 108 | path = tab.routing.currentPath |
| 109 | } |
| 110 | if decodeErr != nil { |
| 111 | markRemoteRuntimeUnknownLocked(tab, path) |
| 112 | a.remoteTabMu.Unlock() |
| 113 | a.emitRuntimeStateChanged() |
| 114 | a.goRemoteTabSafe("remoteRuntimeSync", func() { _, _ = a.SyncRuntimeState() }) |
| 115 | return |
| 116 | } |
| 117 | previous, found := tab.runtimeStates[path] |
| 118 | resync := !found || previous.RuntimeEpoch != state.RuntimeEpoch || (previous.Revision == state.Revision && !reflect.DeepEqual(previous, state)) |
| 119 | if resync { |
| 120 | if reflect.DeepEqual(tab.runtimeConflicts[path], state) { |
| 121 | resync = false |
| 122 | } else { |
| 123 | if tab.runtimeConflicts == nil { |
| 124 | tab.runtimeConflicts = map[string]event.RuntimeStateSnapshot{} |
| 125 | } |
| 126 | tab.runtimeConflicts[path] = state |
| 127 | } |
| 128 | } |
| 129 | changed := acceptRemoteRuntimeStateLocked(tab, path, state, false) |
| 130 | meta := remoteTabMetaLocked(tab) |
| 131 | a.remoteTabMu.Unlock() |
| 132 | if changed { |
| 133 | a.emitRemoteEvent("remote-tab:updated", meta) |
| 134 | a.emitRuntimeStateChanged() |
| 135 | } else if resync { |
| 136 | a.goRemoteTabSafe("remoteRuntimeSync", func() { _, _ = a.SyncRuntimeState() }) |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | type remoteRuntimeTarget struct { |
| 141 | id string |
| 142 | gen, selection uint64 |
| 143 | client *http.Client |
| 144 | states map[string]event.RuntimeStateSnapshot |
| 145 | unknown map[string]uint64 |
| 146 | } |
| 147 | type remoteRuntimeConnection struct { |
| 148 | id, key, base string |
| 149 | client *http.Client |
| 150 | targets []remoteRuntimeTarget |
| 151 | } |
| 152 | |
| 153 | // SyncRuntimeState reconciles each Serve connection once, regardless of how |
| 154 | // many tabs it owns. It is separate from the memory-only snapshot getter. |
| 155 | func (a *App) SyncRuntimeState() (RuntimeStateProjection, error) { |
| 156 | syncer := &a.remoteRuntimeSync |
| 157 | syncer.mu.Lock() |
| 158 | if syncer.pending { |
| 159 | syncer.mu.Unlock() |
| 160 | return a.GetRuntimeStateSnapshot(), nil |
| 161 | } |
| 162 | syncer.pending = true |
| 163 | if syncer.unsupported == nil { |
| 164 | syncer.unsupported = map[string]string{} |
| 165 | } |
| 166 | syncer.mu.Unlock() |
| 167 | defer func() { syncer.mu.Lock(); syncer.pending = false; syncer.mu.Unlock() }() |
| 168 | connections := map[string]remoteRuntimeConnection{} |
| 169 | a.remoteTabMu.Lock() |
| 170 | for id, tab := range a.remoteTabs { |
| 171 | if tab.client == nil || tab.state != "ready" { |
| 172 | continue |
| 173 | } |
| 174 | key := fmt.Sprintf("%s\x00%s\x00%s", tab.ref.HostID, tab.ref.Workspace, tab.base) |
| 175 | conn := connections[key] |
| 176 | conn.id, conn.key, conn.base, conn.client = id, key, tab.base, tab.client |
| 177 | states := map[string]event.RuntimeStateSnapshot{} |
| 178 | maps.Copy(states, tab.runtimeStates) |
| 179 | conn.targets = append(conn.targets, remoteRuntimeTarget{id, tab.gen, tab.selectionRevision, tab.client, states, maps.Clone(tab.runtimeUnknown)}) |
| 180 | connections[key] = conn |
| 181 | } |
| 182 | a.remoteTabMu.Unlock() |
| 183 | var firstErr error |
| 184 | |
| 185 | for _, conn := range connections { |
| 186 | if err := a.syncRemoteRuntimeConnection(conn); err != nil && firstErr == nil { |
| 187 | firstErr = err |
| 188 | } |
| 189 | } |
| 190 | if firstErr != nil { |
| 191 | slog.Warn("remote runtime synchronization failed", "source", "serve", "reason", "reconcile", "connections", len(connections), "failures", remoteRuntimeDiagnostics.failures.Add(1)) |
| 192 | } |
| 193 | a.emitRuntimeStateChanged() |
| 194 | return a.GetRuntimeStateSnapshot(), firstErr |
| 195 | } |
| 196 | |
| 197 | func (a *App) markRemoteRuntimeSyncFailed(targets []remoteRuntimeTarget, failed bool) { |
| 198 | a.remoteTabMu.Lock() |
| 199 | defer a.remoteTabMu.Unlock() |
| 200 | for _, target := range targets { |
| 201 | if tab := a.remoteTabs[target.id]; tab != nil && tab.gen == target.gen && tab.client == target.client && tab.selectionRevision == target.selection { |
| 202 | tab.runtime.syncFailed = failed |
| 203 | if failed { |
| 204 | markRemoteRuntimeUnknownLocked(tab, tab.routing.currentPath) |
| 205 | for path := range tab.runtimeStates { |
| 206 | markRemoteRuntimeUnknownLocked(tab, path) |
| 207 | } |
| 208 | } |
| 209 | } |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | func (a *App) syncRemoteRuntimeConnection(conn remoteRuntimeConnection) error { |
| 214 | syncer := &a.remoteRuntimeSync |
| 215 | sort.Slice(conn.targets, func(i, j int) bool { return conn.targets[i].id < conn.targets[j].id }) |
| 216 | first := conn.targets[0] |
| 217 | capabilityKey := fmt.Sprintf("%s\x00%s\x00%d\x00%p", conn.key, first.id, first.gen, first.client) |
| 218 | syncer.mu.Lock() |
| 219 | unsupported := syncer.unsupported[conn.key] == capabilityKey |
| 220 | syncer.mu.Unlock() |
| 221 | if unsupported { |
| 222 | _, err := a.RemoteTabStatus(conn.id) |
| 223 | a.markRemoteRuntimeSyncFailed(conn.targets, err != nil) |
| 224 | return err |
| 225 | } |
| 226 | ctx, cancel := context.WithTimeout(a.bootContext(), 5*time.Second) |
| 227 | req, err := http.NewRequestWithContext(ctx, http.MethodGet, serveURL(conn.base, "/runtime-states"), nil) |
| 228 | if err != nil { |
| 229 | cancel() |
| 230 | a.markRemoteRuntimeSyncFailed(conn.targets, true) |
| 231 | return err |
| 232 | } |
| 233 | resp, err := conn.client.Do(req) |
| 234 | if err != nil { |
| 235 | cancel() |
| 236 | a.markRemoteRuntimeSyncFailed(conn.targets, true) |
| 237 | return err |
| 238 | } |
| 239 | if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusNotImplemented { |
| 240 | resp.Body.Close() |
| 241 | cancel() |
| 242 | syncer.mu.Lock() |
| 243 | syncer.unsupported[conn.key] = capabilityKey |
| 244 | syncer.mu.Unlock() |
| 245 | _, err := a.RemoteTabStatus(conn.id) |
| 246 | a.markRemoteRuntimeSyncFailed(conn.targets, err != nil) |
| 247 | return err |
| 248 | } |
| 249 | var payload struct { |
| 250 | SchemaVersion int `json:"schemaVersion"` |
| 251 | Sessions []struct { |
| 252 | SessionPath string `json:"sessionPath"` |
| 253 | State json.RawMessage `json:"state"` |
| 254 | } `json:"sessions"` |
| 255 | } |
| 256 | err = json.NewDecoder(http.MaxBytesReader(nil, resp.Body, 4<<20)).Decode(&payload) |
| 257 | resp.Body.Close() |
| 258 | cancel() |
| 259 | if err != nil || resp.StatusCode != http.StatusOK || payload.SchemaVersion != 1 || payload.Sessions == nil { |
| 260 | a.markRemoteRuntimeSyncFailed(conn.targets, true) |
| 261 | return fmt.Errorf("runtime state synchronization failed") |
| 262 | } |
| 263 | states := make(map[string]event.RuntimeStateSnapshot, len(payload.Sessions)) |
| 264 | for _, session := range payload.Sessions { |
| 265 | state, decodeErr := decodeRemoteRuntimeState(session.State) |
| 266 | _, duplicate := states[session.SessionPath] |
| 267 | if decodeErr != nil || duplicate { |
| 268 | a.markRemoteRuntimeSyncFailed(conn.targets, true) |
| 269 | return fmt.Errorf("invalid runtime state synchronization payload") |
| 270 | } |
| 271 | if session.SessionPath == "" { |
| 272 | // An identity session a Serve has not routed yet reports no legacy |
| 273 | // path; skip it instead of failing the whole reconciliation. |
| 274 | continue |
| 275 | } |
| 276 | states[session.SessionPath] = state |
| 277 | } |
| 278 | a.applyRemoteRuntimeSnapshot(conn, states) |
| 279 | return nil |
| 280 | } |
| 281 | |
| 282 | func (a *App) applyRemoteRuntimeSnapshot(conn remoteRuntimeConnection, states map[string]event.RuntimeStateSnapshot) { |
| 283 | a.remoteTabMu.Lock() |
| 284 | defer a.remoteTabMu.Unlock() |
| 285 | for _, target := range conn.targets { |
| 286 | tab := a.remoteTabs[target.id] |
| 287 | if tab == nil || tab.base != conn.base || tab.gen != target.gen || tab.client != target.client || tab.selectionRevision != target.selection || tab.routing.rehydratingPath != "" { |
| 288 | continue |
| 289 | } |
| 290 | tab.runtime.syncFailed = false |
| 291 | seen := map[string]bool{} |
| 292 | for path, state := range states { |
| 293 | seen[path] = true |
| 294 | // A failure or malformed frame observed after this GET started |
| 295 | // requires a fresh request, even when the cached facts are unchanged. |
| 296 | if tab.runtimeUnknown[path] != target.unknown[path] { |
| 297 | continue |
| 298 | } |
| 299 | previous := tab.runtimeStates[path] |
| 300 | // A GET can establish an epoch only if no newer source update |
| 301 | // has changed this binding while it was in flight. |
| 302 | if previous.RuntimeEpoch != state.RuntimeEpoch && !reflect.DeepEqual(previous, target.states[path]) { |
| 303 | continue |
| 304 | } |
| 305 | acceptRemoteRuntimeStateLocked(tab, path, state, true) |
| 306 | } |
| 307 | if !seen[tab.routing.currentPath] && reflect.DeepEqual(tab.runtimeStates[tab.routing.currentPath], target.states[tab.routing.currentPath]) { |
| 308 | markRemoteRuntimeUnknownLocked(tab, tab.routing.currentPath) |
| 309 | } |
| 310 | for path, state := range tab.runtimeStates { |
| 311 | if !seen[path] && path != tab.routing.currentPath && reflect.DeepEqual(state, target.states[path]) && tab.runtimeUnknown[path] == target.unknown[path] { |
| 312 | delete(tab.runtimeStates, path) |
| 313 | delete(tab.runtimeUnknown, path) |
| 314 | delete(tab.routing.running, path) |
| 315 | } |
| 316 | } |
| 317 | } |
| 318 | } |
| 319 | |
| 320 | func markRemoteRuntimeUnknownLocked(tab *remoteTab, path string) { |
| 321 | if tab.runtimeUnknown == nil { |
| 322 | tab.runtimeUnknown = map[string]uint64{} |
| 323 | } |
| 324 | tab.runtime.revision++ |
| 325 | tab.runtimeUnknown[path] = tab.runtime.revision |
| 326 | } |
| 327 |