| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "database/sql" |
| 7 | "encoding" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "io" |
| 12 | "os" |
| 13 | "time" |
| 14 | |
| 15 | "reasonix/internal/historywork" |
| 16 | "reasonix/internal/provider" |
| 17 | ) |
| 18 | |
| 19 | func dagPagerBatch(ctx context.Context, db *sql.DB, p *dagPagerProgress, work func(*sql.Tx) error) error { |
| 20 | tx, err := db.BeginTx(ctx, nil) |
| 21 | if err != nil { |
| 22 | return err |
| 23 | } |
| 24 | defer func() { _ = tx.Rollback() }() |
| 25 | if err := work(tx); err != nil { |
| 26 | return err |
| 27 | } |
| 28 | return commitDAGPagerProgress(ctx, tx, p) |
| 29 | } |
| 30 | |
| 31 | func resumeDAGDisplayPager(ctx context.Context, db *sql.DB, f *os.File, p *dagPagerProgress, source, requestedHead string, observed func(string, int)) (SessionDisplayIndex, string, error) { |
| 32 | st := p.state(source) |
| 33 | p.HeadPositions = make(map[string]int, len(st.headOrder)) |
| 34 | for position, id := range st.headOrder { |
| 35 | p.HeadPositions[id] = position |
| 36 | } |
| 37 | p.Heads = nil |
| 38 | base := p.Offset |
| 39 | decoder := json.NewDecoder(&historywork.Reader{Context: ctx, Source: io.NewSectionReader(f, base, 1<<63-1-base)}) |
| 40 | for p.Phase != "done" { |
| 41 | if err := ctx.Err(); err != nil { |
| 42 | return SessionDisplayIndex{}, "", err |
| 43 | } |
| 44 | phase := p.Phase |
| 45 | err := dagPagerBatch(ctx, db, p, func(tx *sql.Tx) error { |
| 46 | switch phase { |
| 47 | case "scan": |
| 48 | return scanDAGPagerBatch(ctx, tx, f, decoder, base, p, st, requestedHead) |
| 49 | case "chain": |
| 50 | return chainDAGPagerBatch(ctx, tx, p) |
| 51 | case "projection": |
| 52 | return projectDAGPagerBatch(ctx, tx, f, p, st.heads[p.HeadID]) |
| 53 | default: |
| 54 | return ErrSessionDisplayReadModelDamaged |
| 55 | } |
| 56 | }) |
| 57 | if err != nil { |
| 58 | return SessionDisplayIndex{}, "", err |
| 59 | } |
| 60 | if observed != nil { |
| 61 | count := p.Records |
| 62 | switch phase { |
| 63 | case "chain": |
| 64 | count = p.ChainCount |
| 65 | case "projection": |
| 66 | count = p.Header.MessageCount |
| 67 | } |
| 68 | observed(phase, count) |
| 69 | } |
| 70 | } |
| 71 | return p.Header, p.HeadID, ctx.Err() |
| 72 | } |
| 73 | |
| 74 | func scanDAGPagerBatch(ctx context.Context, tx *sql.Tx, f *os.File, decoder *json.Decoder, base int64, p *dagPagerProgress, st *sessionDAGState, requestedHead string) error { |
| 75 | started, from := time.Now(), p.Offset |
| 76 | changed := map[string]bool{SessionMainHead: true} |
| 77 | for range historywork.BatchEntries { |
| 78 | if err := ctx.Err(); err != nil { |
| 79 | return err |
| 80 | } |
| 81 | start := base + decoder.InputOffset() |
| 82 | var e sessionDAGEntry |
| 83 | if err := decoder.Decode(&e); errors.Is(err, io.EOF) { |
| 84 | p.Phase, p.HeadID = "chain", requestedHead |
| 85 | if p.HeadID == "" { |
| 86 | p.HeadID = st.selectedHead() |
| 87 | } |
| 88 | head := st.heads[p.HeadID] |
| 89 | if head == nil { |
| 90 | return fmt.Errorf("history head not found") |
| 91 | } |
| 92 | p.NextID = head.leaf |
| 93 | break |
| 94 | } else if err != nil { |
| 95 | if ctx.Err() != nil { |
| 96 | return ctx.Err() |
| 97 | } |
| 98 | return fmt.Errorf("DAG history: %w", ErrSessionDisplayReadModelDamaged) |
| 99 | } |
| 100 | if e.SchemaVersion != sessionDAGSchemaVersion { |
| 101 | return fmt.Errorf("unsupported DAG schema %d", e.SchemaVersion) |
| 102 | } |
| 103 | p.Offset = base + decoder.InputOffset() |
| 104 | if err := applyDisplayDAGLocation(ctx, tx, f, st, e, start, p.Offset); err != nil { |
| 105 | return err |
| 106 | } |
| 107 | head := e.Head |
| 108 | if head == "" { |
| 109 | head = SessionMainHead |
| 110 | } |
| 111 | changed[head], changed[e.NewHead] = true, true |
| 112 | for p.HeadCount < len(st.headOrder) { |
| 113 | id := st.headOrder[p.HeadCount] |
| 114 | p.HeadPositions[id] = p.HeadCount |
| 115 | changed[id] = true |
| 116 | p.HeadCount++ |
| 117 | } |
| 118 | p.Records++ |
| 119 | if p.Offset-from >= historywork.BatchBytes || time.Since(started) >= historywork.SliceDuration { |
| 120 | break |
| 121 | } |
| 122 | } |
| 123 | p.Selected = st.selected |
| 124 | // Persist only touched heads. Rewriting every head at every parser checkpoint |
| 125 | // would turn a large fork log into quadratic metadata I/O. |
| 126 | for id := range changed { |
| 127 | if h := st.heads[id]; h != nil { |
| 128 | body, err := json.Marshal(snapshotDAGPagerHead(h)) |
| 129 | if err != nil { |
| 130 | return err |
| 131 | } |
| 132 | if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO dag_heads VALUES(?,?,?)`, p.HeadPositions[id], id, body); err != nil { |
| 133 | return err |
| 134 | } |
| 135 | } |
| 136 | } |
| 137 | return nil |
| 138 | } |
| 139 | |
| 140 | func chainDAGPagerBatch(ctx context.Context, tx *sql.Tx, p *dagPagerProgress) error { |
| 141 | started := time.Now() |
| 142 | for count := 0; p.NextID != "" && count < historywork.BatchEntries; count++ { |
| 143 | var parent string |
| 144 | if err := tx.QueryRowContext(ctx, `SELECT parent FROM dag_nodes WHERE id=?`, p.NextID).Scan(&parent); err != nil { |
| 145 | if errors.Is(err, sql.ErrNoRows) { |
| 146 | return fmt.Errorf("DAG history missing ancestor: %w", ErrSessionDisplayReadModelDamaged) |
| 147 | } |
| 148 | return err |
| 149 | } |
| 150 | if _, err := tx.ExecContext(ctx, `INSERT INTO dag_chain VALUES(?,?)`, p.ChainCount, p.NextID); err != nil { |
| 151 | return fmt.Errorf("DAG history cycle or chain write failure: %w", err) |
| 152 | } |
| 153 | p.ChainCount++ |
| 154 | p.NextID = parent |
| 155 | if time.Since(started) >= historywork.SliceDuration { |
| 156 | break |
| 157 | } |
| 158 | } |
| 159 | if p.NextID == "" { |
| 160 | p.Phase, p.NextPosition = "projection", p.ChainCount-1 |
| 161 | var err error |
| 162 | p.Hash, err = sha256.New().(encoding.BinaryMarshaler).MarshalBinary() |
| 163 | return err |
| 164 | } |
| 165 | return nil |
| 166 | } |
| 167 | |
| 168 | func projectDAGPagerBatch(ctx context.Context, tx *sql.Tx, f *os.File, p *dagPagerProgress, head *sessionDAGHead) error { |
| 169 | w := &displayDAGProjectionWriter{ctx: ctx, db: tx, file: f, index: p.Header, users: p.Users, hasher: sha256.New()} |
| 170 | if err := w.hasher.(encoding.BinaryUnmarshaler).UnmarshalBinary(p.Hash); err != nil { |
| 171 | return err |
| 172 | } |
| 173 | started := time.Now() |
| 174 | for count := 0; p.NextPosition >= 0 && count < historywork.BatchEntries; count++ { |
| 175 | if err := appendDAGPagerPosition(ctx, tx, p, head, w); err != nil { |
| 176 | return err |
| 177 | } |
| 178 | p.NextPosition-- |
| 179 | if time.Since(started) >= historywork.SliceDuration { |
| 180 | break |
| 181 | } |
| 182 | } |
| 183 | if p.ChainCount == 0 && head.system != nil { |
| 184 | loc, err := displayDAGSystem(ctx, tx, head.system.ID) |
| 185 | if err != nil { |
| 186 | return err |
| 187 | } |
| 188 | if err := w.append(loc); err != nil { |
| 189 | return err |
| 190 | } |
| 191 | } |
| 192 | p.Header, p.Users = w.index, w.users |
| 193 | var err error |
| 194 | p.Hash, err = w.hasher.(encoding.BinaryMarshaler).MarshalBinary() |
| 195 | if p.NextPosition < 0 { |
| 196 | p.Phase = "done" |
| 197 | p.Header.ContentDigest = fmt.Sprintf("%x", w.hasher.Sum(nil)) |
| 198 | } |
| 199 | return err |
| 200 | } |
| 201 | |
| 202 | func appendDAGPagerPosition(ctx context.Context, tx *sql.Tx, p *dagPagerProgress, head *sessionDAGHead, w *displayDAGProjectionWriter) error { |
| 203 | var id string |
| 204 | var raw []byte |
| 205 | if err := tx.QueryRowContext(ctx, `SELECT n.id,COALESCE(r.location,p.location,n.location) FROM dag_chain c JOIN dag_nodes n ON n.id=c.id |
| 206 | LEFT JOIN dag_overlays p ON p.kind='patch' AND p.id=n.id LEFT JOIN dag_overlays r ON r.kind='redact' AND r.id=n.id WHERE c.position=?`, p.NextPosition).Scan(&id, &raw); err != nil { |
| 207 | return err |
| 208 | } |
| 209 | var loc displayDAGLocation |
| 210 | if err := json.Unmarshal(raw, &loc); err != nil { |
| 211 | return err |
| 212 | } |
| 213 | loc.ID = id |
| 214 | if p.NextPosition == p.ChainCount-1 && head.system != nil { |
| 215 | m, err := readDisplayDAGMessage(ctx, w.file, loc) |
| 216 | if err != nil { |
| 217 | return err |
| 218 | } |
| 219 | sys, err := displayDAGSystem(ctx, tx, head.system.ID) |
| 220 | if err != nil { |
| 221 | return err |
| 222 | } |
| 223 | if m.Role == provider.RoleSystem { |
| 224 | sys.ID = id |
| 225 | loc = sys |
| 226 | } else if err := w.append(sys); err != nil { |
| 227 | return err |
| 228 | } |
| 229 | } |
| 230 | return w.append(loc) |
| 231 | } |
| 232 |