| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "cmp" |
| 5 | "fmt" |
| 6 | "os" |
| 7 | "slices" |
| 8 | "time" |
| 9 | |
| 10 | "reasonix/internal/fileutil" |
| 11 | "reasonix/internal/provider" |
| 12 | "reasonix/internal/store" |
| 13 | ) |
| 14 | |
| 15 | // sessionDAGWriterQuietPeriod is how long another writer must have been silent |
| 16 | // before this process may rewrite the log under it. |
| 17 | const sessionDAGWriterQuietPeriod = 60 * time.Second |
| 18 | |
| 19 | // SessionRotationDeniedError reports why a log could not be rotated or |
| 20 | // upgraded right now; the log simply keeps growing until the condition clears. |
| 21 | type SessionRotationDeniedError struct { |
| 22 | Reason string |
| 23 | } |
| 24 | |
| 25 | func (e *SessionRotationDeniedError) Error() string { |
| 26 | return "session log rotation denied: " + e.Reason |
| 27 | } |
| 28 | |
| 29 | // sessionDAGSingleWriterProof is the precondition for any rewrite of a schema-2 |
| 30 | // log: this runtime holds the lease, no handoff is reserved, and no other |
| 31 | // writer has appended within the quiet period. The caller holds the file lock. |
| 32 | func sessionDAGSingleWriterProof(sessionPath string, st *sessionDAGState, now time.Time) error { |
| 33 | if !SessionLeaseHeldByCurrentRuntime(sessionPath) { |
| 34 | return &SessionRotationDeniedError{Reason: "session lease is not held by this runtime"} |
| 35 | } |
| 36 | if info, err := LoadSessionLeaseInfo(sessionPath); err == nil && handoffReservationActive(info, now) { |
| 37 | return &SessionRotationDeniedError{Reason: "lease handoff reservation is active"} |
| 38 | } |
| 39 | me := SessionWriterID() |
| 40 | for id, w := range st.writers { |
| 41 | if id == me || id == "" { |
| 42 | continue |
| 43 | } |
| 44 | if now.Sub(w.lastActivity) < sessionDAGWriterQuietPeriod { |
| 45 | return &SessionRotationDeniedError{Reason: fmt.Sprintf("writer %s appended %s ago", id, now.Sub(w.lastActivity).Round(time.Second))} |
| 46 | } |
| 47 | } |
| 48 | return nil |
| 49 | } |
| 50 | |
| 51 | // sessionDAGLogOversized bounds a schema-2 log at the schema-1 growth factor |
| 52 | // over the encoded size of every live head's chain. |
| 53 | func sessionDAGLogOversized(st *sessionDAGState) bool { |
| 54 | live := int64(0) |
| 55 | for _, id := range st.liveHeads() { |
| 56 | msgs, _ := st.materialize(id) |
| 57 | if _, size, err := digestAndSizeSessionMessages(msgs); err == nil { |
| 58 | live += size |
| 59 | } |
| 60 | } |
| 61 | return sessionEventLogOversized(st.size, live) |
| 62 | } |
| 63 | |
| 64 | // rotateSessionDAG atomically replaces the log with the next generation: live |
| 65 | // heads and their chains keep their ids, patches are folded, redactions are |
| 66 | // applied physically, and everything unreachable is dropped. The caller must |
| 67 | // hold the file lock and have obtained the single-writer proof. |
| 68 | func rotateSessionDAG(sessionPath string, st *sessionDAGState, now time.Time) error { |
| 69 | path := store.SessionEventLog(sessionPath) |
| 70 | if path == "" { |
| 71 | return fmt.Errorf("empty session event log path") |
| 72 | } |
| 73 | entries, err := buildRotatedSessionDAG(st, now) |
| 74 | if err != nil { |
| 75 | return err |
| 76 | } |
| 77 | data, err := encodeSessionDAGEntries(entries, now) |
| 78 | if err != nil { |
| 79 | return err |
| 80 | } |
| 81 | fileutil.Crash("dag-rotate", path) |
| 82 | staged, err := fileutil.StageAtomicWrite(path, data, 0o600) |
| 83 | if err != nil { |
| 84 | return err |
| 85 | } |
| 86 | // The marker brackets the window an unlocked appender cannot see into: |
| 87 | // bytes it lands before the late read are carried, bytes after it are |
| 88 | // re-appended by the appender once the marker clears. |
| 89 | marker := store.SessionEventLogRotating(sessionPath) |
| 90 | if err := os.WriteFile(marker, nil, 0o600); err != nil { |
| 91 | _ = os.Remove(staged) |
| 92 | return err |
| 93 | } |
| 94 | defer os.Remove(marker) |
| 95 | if hook := sessionDAGRotateBeforeReplace; hook != nil { |
| 96 | hook(sessionPath) |
| 97 | } |
| 98 | if err := appendLateLinesToStaged(path, st.size, staged); err != nil { |
| 99 | _ = os.Remove(staged) |
| 100 | return err |
| 101 | } |
| 102 | return fileutil.PublishStagedWrite(staged, path) |
| 103 | } |
| 104 | |
| 105 | func buildRotatedSessionDAG(st *sessionDAGState, now time.Time) ([]sessionDAGEntry, error) { |
| 106 | keep := st.reachable() |
| 107 | live := st.liveHeads() |
| 108 | selected := st.selectedHead() |
| 109 | |
| 110 | var dropped, tombstones []string |
| 111 | for id := range st.nodes { |
| 112 | if _, ok := keep[id]; !ok { |
| 113 | dropped = append(dropped, id) |
| 114 | } |
| 115 | } |
| 116 | for id := range st.redactions { |
| 117 | if _, ok := keep[id]; ok { |
| 118 | tombstones = append(tombstones, id) |
| 119 | } |
| 120 | } |
| 121 | slices.Sort(dropped) |
| 122 | slices.Sort(tombstones) |
| 123 | |
| 124 | heads := make([]SessionHead, 0, len(live)) |
| 125 | for _, id := range live { |
| 126 | heads = append(heads, st.headRecord(id, selected)) |
| 127 | } |
| 128 | entries := []sessionDAGEntry{ |
| 129 | {Type: sessionDAGTypeLog, At: now, Generation: st.generation + 1, RotatedFrom: st.generation, UpgradedFrom: st.upgradedFrom}, |
| 130 | {Type: sessionDAGTypeCheckpoint, At: now, SelectedHead: selected, Heads: heads, Dropped: dropped, Tombstones: tombstones}, |
| 131 | } |
| 132 | |
| 133 | writers := make([]*sessionDAGWriter, 0, len(st.writers)) |
| 134 | for _, w := range st.writers { |
| 135 | writers = append(writers, w) |
| 136 | } |
| 137 | slices.SortFunc(writers, func(a, b *sessionDAGWriter) int { return cmp.Compare(a.id, b.id) }) |
| 138 | for _, w := range writers { |
| 139 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeWriter, Writer: w.id, At: w.lastActivity, PID: w.pid, Hostname: w.hostname, LeaseGeneration: w.leaseGeneration}) |
| 140 | } |
| 141 | |
| 142 | forks := make([]*sessionDAGHead, 0, len(live)) |
| 143 | for _, id := range live { |
| 144 | if id != SessionMainHead { |
| 145 | forks = append(forks, st.heads[id]) |
| 146 | } |
| 147 | } |
| 148 | slices.SortStableFunc(forks, func(a, b *sessionDAGHead) int { |
| 149 | if c := a.createdAt.Compare(b.createdAt); c != 0 { |
| 150 | return c |
| 151 | } |
| 152 | return cmp.Compare(a.id, b.id) |
| 153 | }) |
| 154 | for _, h := range forks { |
| 155 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeFork, Head: h.parentHead, NewHead: h.id, From: h.forkFrom, Kind: h.kind, Name: h.name, Writer: h.writer, At: h.createdAt}) |
| 156 | } |
| 157 | if main := st.heads[SessionMainHead]; main != nil && main.name != "" && !main.retired { |
| 158 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeRename, Head: SessionMainHead, Name: main.name, At: main.createdAt}) |
| 159 | } |
| 160 | for _, id := range live { |
| 161 | h := st.heads[id] |
| 162 | if h.system == nil { |
| 163 | continue |
| 164 | } |
| 165 | raw, err := encodeSessionDAGMessage(*h.system) |
| 166 | if err != nil { |
| 167 | return nil, err |
| 168 | } |
| 169 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeSystem, Head: id, Msgs: raw, At: h.createdAt}) |
| 170 | } |
| 171 | |
| 172 | nodes := make([]*sessionDAGNode, 0, len(keep)) |
| 173 | for id := range keep { |
| 174 | nodes = append(nodes, st.nodes[id]) |
| 175 | } |
| 176 | slices.SortFunc(nodes, func(a, b *sessionDAGNode) int { return cmp.Compare(a.offset, b.offset) }) |
| 177 | digests := map[string]string{} |
| 178 | for _, n := range nodes { |
| 179 | m := st.appliedMessage(n) |
| 180 | parentDigest := "" |
| 181 | if _, ok := keep[n.parent]; ok { |
| 182 | parentDigest = digests[n.parent] |
| 183 | } |
| 184 | e, err := newSessionDAGMessageEntry(n.head, n.parent, parentDigest, n.turn, m, n.at) |
| 185 | if err != nil { |
| 186 | return nil, err |
| 187 | } |
| 188 | if _, ok := keep[n.parent]; !ok { |
| 189 | e.Parent = "" |
| 190 | } |
| 191 | e.Writer = n.writer |
| 192 | digests[n.id] = e.Digest |
| 193 | entries = append(entries, e) |
| 194 | } |
| 195 | |
| 196 | for _, id := range live { |
| 197 | h := st.heads[id] |
| 198 | if c := h.compaction; c != nil { |
| 199 | if _, ok := keep[c.coveredLeaf]; ok { |
| 200 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeCompaction, Head: id, CoveredLeaf: c.coveredLeaf, CoveredCount: c.coveredCount, PrefixHash: c.prefixHash, At: c.at}) |
| 201 | } |
| 202 | } |
| 203 | if t := h.openTurn; t != nil { |
| 204 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeTurnBegin, Head: id, Turn: t.turn, Leaf: t.leaf, PreserveUser: t.preserveUser, Writer: t.writer, At: t.at}) |
| 205 | } |
| 206 | } |
| 207 | if h := st.heads[st.selected]; h != nil && !h.retired { |
| 208 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeSelect, Head: st.selected, Reason: "rotation", At: now}) |
| 209 | } |
| 210 | return entries, nil |
| 211 | } |
| 212 | |
| 213 | // upgradeSessionLogToDAG replaces a schema-1 log with generation 1 of the |
| 214 | // schema-2 log: one message entry per transcript message under the main head, |
| 215 | // keeping the ids the caller already assigned, plus an open turn marker when |
| 216 | // the schema-1 in-flight sidecar recorded one. The caller holds the file lock |
| 217 | // and has obtained the single-writer proof. |
| 218 | func upgradeSessionLogToDAG(sessionPath string, msgs []provider.Message, times []time.Time, inFlight *InFlightTurnMeta, now time.Time) error { |
| 219 | path := store.SessionEventLog(sessionPath) |
| 220 | if path == "" { |
| 221 | return fmt.Errorf("empty session event log path") |
| 222 | } |
| 223 | host, _ := os.Hostname() |
| 224 | entries := []sessionDAGEntry{ |
| 225 | {Type: sessionDAGTypeLog, At: now, Generation: 1, UpgradedFrom: sessionEventSchemaVersion}, |
| 226 | {Type: sessionDAGTypeWriter, At: now, PID: os.Getpid(), Hostname: host}, |
| 227 | } |
| 228 | parent, digest := "", "" |
| 229 | for i, m := range msgs { |
| 230 | if m.ID == "" { |
| 231 | return fmt.Errorf("upgrade session log: message %d has no id", i) |
| 232 | } |
| 233 | at := now |
| 234 | if i < len(times) && !times[i].IsZero() { |
| 235 | at = times[i] |
| 236 | } |
| 237 | e, err := newSessionDAGMessageEntry(SessionMainHead, parent, digest, "", m, at) |
| 238 | if err != nil { |
| 239 | return err |
| 240 | } |
| 241 | entries = append(entries, e) |
| 242 | parent, digest = m.ID, e.Digest |
| 243 | } |
| 244 | if inFlight != nil && inFlight.ID != "" && inFlight.StartMessageIndex >= 0 && inFlight.StartMessageIndex <= len(msgs) { |
| 245 | leaf := "" |
| 246 | if inFlight.StartMessageIndex > 0 { |
| 247 | leaf = msgs[inFlight.StartMessageIndex-1].ID |
| 248 | } |
| 249 | at := inFlight.StartedAt |
| 250 | if at.IsZero() { |
| 251 | at = now |
| 252 | } |
| 253 | entries = append(entries, sessionDAGEntry{Type: sessionDAGTypeTurnBegin, Head: SessionMainHead, Turn: inFlight.ID, Leaf: leaf, PreserveUser: inFlight.PreserveUser, At: at}) |
| 254 | } |
| 255 | data, err := encodeSessionDAGEntries(entries, now) |
| 256 | if err != nil { |
| 257 | return err |
| 258 | } |
| 259 | fileutil.Crash("dag-upgrade", path) |
| 260 | return fileutil.AtomicWriteFileStrict(path, data, 0o600) |
| 261 | } |
| 262 |