| 1 | package sessioncatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "errors" |
| 7 | "maps" |
| 8 | "runtime" |
| 9 | "strings" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | ) |
| 14 | |
| 15 | const repairWakeKey = "session-catalog-repair-wake" |
| 16 | |
| 17 | const repairClaimSQL = `UPDATE catalog_sessions SET repair_state='active',repair_attempts=?, |
| 18 | repair_retry_at=?,repair_error_kind='' WHERE path_key=? AND turns_state='unknown' |
| 19 | AND repair_state=? AND repair_attempts=? AND repair_retry_at=? AND repair_source_fingerprint=?` |
| 20 | |
| 21 | const repairResetSQL = `UPDATE catalog_sessions SET repair_state='pending',repair_attempts=0, |
| 22 | repair_retry_at=0,repair_error_kind='',content_fingerprint=?,meta_fingerprint=?,repair_source_fingerprint=?, |
| 23 | repair_engine_version=? WHERE path_key=? AND turns_state='unknown' AND repair_state='active' |
| 24 | AND repair_attempts=? AND repair_retry_at=? AND repair_source_fingerprint=?` |
| 25 | |
| 26 | const repairCompleteSQL = `UPDATE catalog_sessions SET preview=?,turns=?,turns_state='valid',health='ok', |
| 27 | content_fingerprint=?,meta_fingerprint=?,repair_state='complete',repair_attempts=0,repair_retry_at=0, |
| 28 | repair_error_kind='',repair_source_fingerprint=?,repair_engine_version=? WHERE path_key=? AND turns_state='unknown' |
| 29 | AND repair_state='active' AND repair_attempts=? AND repair_retry_at=? AND repair_source_fingerprint=?` |
| 30 | |
| 31 | const repairDeferredSQL = `UPDATE catalog_sessions SET health=?,repair_state=?,repair_attempts=?,repair_retry_at=?, |
| 32 | repair_error_kind=?,content_fingerprint=?,meta_fingerprint=?,repair_source_fingerprint=?,repair_engine_version=? |
| 33 | WHERE path_key=? AND turns_state='unknown' AND repair_state='active' AND repair_attempts=? |
| 34 | AND repair_retry_at=? AND repair_source_fingerprint=?` |
| 35 | |
| 36 | type repairItem struct { |
| 37 | path string |
| 38 | pathKey string |
| 39 | target DirectoryTarget |
| 40 | topicID string |
| 41 | workspaceRootKey string |
| 42 | attempts int |
| 43 | state string |
| 44 | retryAt int64 |
| 45 | sourceFingerprint string |
| 46 | } |
| 47 | |
| 48 | type repairOutcome struct { |
| 49 | item repairItem |
| 50 | result agent.SessionListingRepairResult |
| 51 | err error |
| 52 | } |
| 53 | |
| 54 | type repairBatchStatements struct { |
| 55 | reset *sql.Stmt |
| 56 | complete *sql.Stmt |
| 57 | deferred *sql.Stmt |
| 58 | } |
| 59 | |
| 60 | func prepareRepairBatchStatements(ctx context.Context, tx *sql.Tx) (repairBatchStatements, error) { |
| 61 | var statements repairBatchStatements |
| 62 | var err error |
| 63 | if statements.reset, err = tx.PrepareContext(ctx, repairResetSQL); err != nil { |
| 64 | return statements, err |
| 65 | } |
| 66 | if statements.complete, err = tx.PrepareContext(ctx, repairCompleteSQL); err != nil { |
| 67 | statements.close() |
| 68 | return statements, err |
| 69 | } |
| 70 | if statements.deferred, err = tx.PrepareContext(ctx, repairDeferredSQL); err != nil { |
| 71 | statements.close() |
| 72 | return statements, err |
| 73 | } |
| 74 | return statements, nil |
| 75 | } |
| 76 | |
| 77 | func (s repairBatchStatements) close() { |
| 78 | for _, statement := range []*sql.Stmt{s.reset, s.complete, s.deferred} { |
| 79 | if statement != nil { |
| 80 | _ = statement.Close() |
| 81 | } |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | func (c *Catalog) enqueueRepair(path string) { |
| 86 | if c == nil || c.opts.DisableRepair || strings.TrimSpace(path) == "" { |
| 87 | return |
| 88 | } |
| 89 | if _, loaded := c.repairQueued.LoadOrStore(repairWakeKey, struct{}{}); loaded { |
| 90 | return |
| 91 | } |
| 92 | select { |
| 93 | case c.repairCh <- path: |
| 94 | case <-c.stop: |
| 95 | c.repairQueued.Delete(repairWakeKey) |
| 96 | default: |
| 97 | c.repairQueued.Delete(repairWakeKey) |
| 98 | } |
| 99 | } |
| 100 | |
| 101 | func (c *Catalog) enqueuePersistedRepairs(ctx context.Context) { |
| 102 | if ctx.Err() == nil { |
| 103 | c.enqueueRepair(repairWakeKey) |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | func (c *Catalog) repairLoop() { |
| 108 | defer c.workers.Done() |
| 109 | timer := time.NewTimer(0) |
| 110 | defer timer.Stop() |
| 111 | for { |
| 112 | select { |
| 113 | case <-c.repairCh: |
| 114 | c.repairQueued.Delete(repairWakeKey) |
| 115 | c.runRepairWave(c.workerCtx) |
| 116 | resetRepairTimer(timer, c.nextRepairDelay(c.workerCtx)) |
| 117 | case <-timer.C: |
| 118 | c.runRepairWave(c.workerCtx) |
| 119 | resetRepairTimer(timer, c.nextRepairDelay(c.workerCtx)) |
| 120 | case <-c.stop: |
| 121 | return |
| 122 | } |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | func resetRepairTimer(timer *time.Timer, delay time.Duration) { |
| 127 | if delay < 0 { |
| 128 | delay = 0 |
| 129 | } |
| 130 | if !timer.Stop() { |
| 131 | select { |
| 132 | case <-timer.C: |
| 133 | default: |
| 134 | } |
| 135 | } |
| 136 | timer.Reset(delay) |
| 137 | } |
| 138 | |
| 139 | func (c *Catalog) nextRepairDelay(ctx context.Context) time.Duration { |
| 140 | if ctx.Err() != nil { |
| 141 | return time.Hour |
| 142 | } |
| 143 | var next sql.NullInt64 |
| 144 | err := c.db.QueryRowContext(ctx, `SELECT MIN(repair_retry_at) FROM catalog_sessions |
| 145 | WHERE turns_state='unknown' AND repair_state IN ('pending','deferred','active')`).Scan(&next) |
| 146 | if err != nil || !next.Valid { |
| 147 | return time.Hour |
| 148 | } |
| 149 | delay := time.UnixMilli(next.Int64).Sub(c.opts.Now()) |
| 150 | if delay < 250*time.Millisecond { |
| 151 | // A failed claim must not turn an already-due row into a timer hot loop. |
| 152 | // Fresh source writes still wake repairCh immediately. |
| 153 | return 250 * time.Millisecond |
| 154 | } |
| 155 | return delay |
| 156 | } |
| 157 | |
| 158 | func (c *Catalog) resetRepairSchedule(ctx context.Context) error { |
| 159 | _, err := c.db.ExecContext(ctx, `UPDATE catalog_sessions SET |
| 160 | repair_state=CASE WHEN turns_state='unknown' THEN 'pending' ELSE 'complete' END, |
| 161 | repair_attempts=0, |
| 162 | repair_retry_at=CASE WHEN turns_state='unknown' THEN 0 ELSE repair_retry_at END, |
| 163 | repair_error_kind='', repair_engine_version=? |
| 164 | WHERE repair_engine_version<>?`, repairEngineVersion, repairEngineVersion) |
| 165 | return err |
| 166 | } |
| 167 | |
| 168 | func (c *Catalog) runRepairWave(workerCtx context.Context) { |
| 169 | if workerCtx.Err() != nil { |
| 170 | return |
| 171 | } |
| 172 | started := c.opts.Now() |
| 173 | processed := false |
| 174 | dirty := map[string]DirectoryTarget{} |
| 175 | failed := false |
| 176 | for workerCtx.Err() == nil && !failed { |
| 177 | items, err := c.claimDueRepairs(workerCtx, 64) |
| 178 | if err != nil || len(items) == 0 { |
| 179 | break |
| 180 | } |
| 181 | processed = true |
| 182 | batch := make([]repairOutcome, 0, len(items)) |
| 183 | batchStarted := c.opts.Now() |
| 184 | for _, item := range items { |
| 185 | ctx, cancel := context.WithTimeout(workerCtx, 30*time.Second) |
| 186 | var result agent.SessionListingRepairResult |
| 187 | var repairErr error |
| 188 | if c.testRepairSessionHook != nil { |
| 189 | result, repairErr = c.testRepairSessionHook(ctx, item.path) |
| 190 | } else { |
| 191 | result, repairErr = agent.RepairSessionListingProjection(ctx, item.path) |
| 192 | } |
| 193 | cancel() |
| 194 | if workerCtx.Err() != nil { |
| 195 | return |
| 196 | } |
| 197 | batch = append(batch, repairOutcome{item: item, result: result, err: repairErr}) |
| 198 | if len(batch) >= 64 || c.opts.Now().Sub(batchStarted) >= 250*time.Millisecond { |
| 199 | if err := c.applyRepairBatch(workerCtx, batch, dirty); err != nil { |
| 200 | failed = true |
| 201 | break |
| 202 | } |
| 203 | batch = batch[:0] |
| 204 | batchStarted = c.opts.Now() |
| 205 | } |
| 206 | runtime.Gosched() |
| 207 | } |
| 208 | if !failed { |
| 209 | failed = c.applyRepairBatch(workerCtx, batch, dirty) != nil |
| 210 | } |
| 211 | } |
| 212 | for _, target := range dirty { |
| 213 | c.RequestReconcile(target) |
| 214 | } |
| 215 | if processed { |
| 216 | c.statusMu.Lock() |
| 217 | c.status.LastRepairDurationMS = max(int64(0), c.opts.Now().Sub(started).Milliseconds()) |
| 218 | c.statusMu.Unlock() |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | func (c *Catalog) claimDueRepairs(ctx context.Context, limit int) ([]repairItem, error) { |
| 223 | c.mutationMu.Lock() |
| 224 | defer c.mutationMu.Unlock() |
| 225 | tx, err := c.db.BeginTx(ctx, nil) |
| 226 | if err != nil { |
| 227 | return nil, err |
| 228 | } |
| 229 | now := c.opts.Now() |
| 230 | rows, err := tx.QueryContext(ctx, `SELECT path,path_key,directory,scope,workspace_root,workspace_root_key,topic_id, |
| 231 | repair_attempts,repair_state,repair_retry_at,repair_source_fingerprint |
| 232 | FROM catalog_sessions WHERE turns_state='unknown' |
| 233 | AND repair_state IN ('pending','deferred','active') AND repair_retry_at<=? |
| 234 | ORDER BY repair_retry_at ASC,last_activity_at DESC,path_key ASC LIMIT ?`, now.UnixMilli(), limit) |
| 235 | if err != nil { |
| 236 | _ = tx.Rollback() |
| 237 | return nil, err |
| 238 | } |
| 239 | var items []repairItem |
| 240 | for rows.Next() { |
| 241 | var item repairItem |
| 242 | if err := rows.Scan(&item.path, &item.pathKey, &item.target.Path, &item.target.Scope, |
| 243 | &item.target.WorkspaceRoot, &item.workspaceRootKey, &item.topicID, &item.attempts, &item.state, |
| 244 | &item.retryAt, &item.sourceFingerprint); err != nil { |
| 245 | _ = rows.Close() |
| 246 | _ = tx.Rollback() |
| 247 | return nil, err |
| 248 | } |
| 249 | items = append(items, item) |
| 250 | } |
| 251 | if err := rows.Close(); err != nil { |
| 252 | _ = tx.Rollback() |
| 253 | return nil, err |
| 254 | } |
| 255 | claimStmt, err := tx.PrepareContext(ctx, repairClaimSQL) |
| 256 | if err != nil { |
| 257 | _ = tx.Rollback() |
| 258 | return nil, err |
| 259 | } |
| 260 | defer claimStmt.Close() |
| 261 | claimed := items[:0] |
| 262 | for _, item := range items { |
| 263 | claimedItem, ok, err := claimRepairItem(ctx, claimStmt, item, now) |
| 264 | if err != nil { |
| 265 | _ = tx.Rollback() |
| 266 | return nil, err |
| 267 | } |
| 268 | if ok { |
| 269 | claimed = append(claimed, claimedItem) |
| 270 | } |
| 271 | } |
| 272 | if err := tx.Commit(); err != nil { |
| 273 | return nil, err |
| 274 | } |
| 275 | c.refreshCounts(ctx) |
| 276 | return claimed, nil |
| 277 | } |
| 278 | |
| 279 | func claimRepairItem(ctx context.Context, claimStmt *sql.Stmt, item repairItem, now time.Time) (repairItem, bool, error) { |
| 280 | previousState, previousAttempts, previousRetryAt := item.state, item.attempts, item.retryAt |
| 281 | lease := 30 * time.Second |
| 282 | if item.state == "active" { |
| 283 | item.attempts++ |
| 284 | lease = repairBackoff(item.attempts + 1) |
| 285 | } |
| 286 | item.state = "active" |
| 287 | item.retryAt = now.Add(lease).UnixMilli() |
| 288 | result, err := claimStmt.ExecContext(ctx, |
| 289 | item.attempts, item.retryAt, item.pathKey, previousState, previousAttempts, previousRetryAt, item.sourceFingerprint) |
| 290 | if err != nil { |
| 291 | return repairItem{}, false, err |
| 292 | } |
| 293 | rows, err := result.RowsAffected() |
| 294 | if err != nil { |
| 295 | return repairItem{}, false, err |
| 296 | } |
| 297 | return item, rows == 1, nil |
| 298 | } |
| 299 | |
| 300 | func (c *Catalog) applyRepairBatch(ctx context.Context, outcomes []repairOutcome, dirty map[string]DirectoryTarget) error { |
| 301 | if len(outcomes) == 0 || ctx.Err() != nil { |
| 302 | return ctx.Err() |
| 303 | } |
| 304 | generations, releaseGenerations, err := lockRepairBatchGenerations(ctx, outcomes) |
| 305 | if err != nil { |
| 306 | return err |
| 307 | } |
| 308 | released := false |
| 309 | release := func() { |
| 310 | if !released { |
| 311 | releaseGenerations() |
| 312 | released = true |
| 313 | } |
| 314 | } |
| 315 | defer release() |
| 316 | return c.applyGuardedRepairBatch(ctx, outcomes, generations, dirty, release) |
| 317 | } |
| 318 | |
| 319 | func (c *Catalog) applyGuardedRepairBatch(ctx context.Context, outcomes []repairOutcome, generations []repairBatchGeneration, |
| 320 | dirty map[string]DirectoryTarget, releaseGenerations func()) error { |
| 321 | c.mutationMu.Lock() |
| 322 | if err := c.repairBatchTestError("begin"); err != nil { |
| 323 | c.mutationMu.Unlock() |
| 324 | return err |
| 325 | } |
| 326 | tx, err := c.db.BeginTx(ctx, nil) |
| 327 | if err != nil { |
| 328 | c.mutationMu.Unlock() |
| 329 | return err |
| 330 | } |
| 331 | rollback := func(err error) error { |
| 332 | _ = tx.Rollback() |
| 333 | c.mutationMu.Unlock() |
| 334 | return err |
| 335 | } |
| 336 | statements, err := prepareRepairBatchStatements(ctx, tx) |
| 337 | if err != nil { |
| 338 | return rollback(err) |
| 339 | } |
| 340 | defer statements.close() |
| 341 | affected := map[TopicKey]struct{}{} |
| 342 | roots := map[string]struct{}{} |
| 343 | committedDirty := map[string]DirectoryTarget{} |
| 344 | mutated := 0 |
| 345 | for index, outcome := range outcomes { |
| 346 | if err := c.repairBatchTestError("update"); err != nil { |
| 347 | return rollback(err) |
| 348 | } |
| 349 | contentFingerprint, metaFingerprint := repairBatchFingerprints(outcome, generations[index]) |
| 350 | sourceFingerprint := contentFingerprint + "\x00" + metaFingerprint |
| 351 | resultHasGeneration := outcome.result.ContentFingerprint != "" || outcome.result.MetaFingerprint != "" |
| 352 | resultSourceFingerprint := outcome.result.ContentFingerprint + "\x00" + outcome.result.MetaFingerprint |
| 353 | if generations[index].locked && resultHasGeneration && resultSourceFingerprint != sourceFingerprint { |
| 354 | result, updateErr := statements.reset.ExecContext(ctx, |
| 355 | contentFingerprint, metaFingerprint, sourceFingerprint, repairEngineVersion, outcome.item.pathKey, |
| 356 | outcome.item.attempts, outcome.item.retryAt, outcome.item.sourceFingerprint) |
| 357 | if updateErr != nil { |
| 358 | return rollback(updateErr) |
| 359 | } |
| 360 | rows, rowsErr := result.RowsAffected() |
| 361 | if rowsErr != nil { |
| 362 | return rollback(rowsErr) |
| 363 | } |
| 364 | if rows == 1 { |
| 365 | mutated++ |
| 366 | roots[outcome.item.target.WorkspaceRoot] = struct{}{} |
| 367 | } |
| 368 | continue |
| 369 | } |
| 370 | state, attempts, retryAt, errorKind, health := repairDisposition(outcome, c.opts.Now()) |
| 371 | var updateResult sql.Result |
| 372 | if state == "complete" { |
| 373 | updateResult, err = statements.complete.ExecContext(ctx, |
| 374 | outcome.result.Preview, outcome.result.Turns, contentFingerprint, metaFingerprint, |
| 375 | sourceFingerprint, repairEngineVersion, outcome.item.pathKey, outcome.item.attempts, |
| 376 | outcome.item.retryAt, outcome.item.sourceFingerprint) |
| 377 | } else { |
| 378 | updateResult, err = statements.deferred.ExecContext(ctx, health, state, attempts, retryAt, errorKind, |
| 379 | contentFingerprint, metaFingerprint, sourceFingerprint, repairEngineVersion, outcome.item.pathKey, |
| 380 | outcome.item.attempts, outcome.item.retryAt, outcome.item.sourceFingerprint) |
| 381 | } |
| 382 | if err != nil { |
| 383 | return rollback(err) |
| 384 | } |
| 385 | updatedRows, rowsErr := updateResult.RowsAffected() |
| 386 | if rowsErr != nil { |
| 387 | return rollback(rowsErr) |
| 388 | } |
| 389 | if updatedRows == 0 { |
| 390 | continue |
| 391 | } |
| 392 | mutated++ |
| 393 | if state == "complete" { |
| 394 | committedDirty[queuePathKey(outcome.item.target.Path)] = outcome.item.target |
| 395 | } |
| 396 | if outcome.item.topicID != "" { |
| 397 | affected[TopicKey{Scope: outcome.item.target.Scope, WorkspaceRoot: outcome.item.target.WorkspaceRoot, |
| 398 | workspaceKey: outcome.item.workspaceRootKey, TopicID: outcome.item.topicID}] = struct{}{} |
| 399 | } |
| 400 | roots[outcome.item.target.WorkspaceRoot] = struct{}{} |
| 401 | } |
| 402 | if mutated == 0 { |
| 403 | _ = tx.Rollback() |
| 404 | c.mutationMu.Unlock() |
| 405 | return nil |
| 406 | } |
| 407 | for key := range affected { |
| 408 | if err := c.recomputeTopic(ctx, tx, key); err != nil { |
| 409 | return rollback(err) |
| 410 | } |
| 411 | } |
| 412 | if err := c.repairBatchTestError("revision"); err != nil { |
| 413 | return rollback(err) |
| 414 | } |
| 415 | revision, err := bumpRevision(ctx, tx) |
| 416 | if err != nil { |
| 417 | return rollback(err) |
| 418 | } |
| 419 | if err := c.repairBatchTestError("commit"); err != nil { |
| 420 | return rollback(err) |
| 421 | } |
| 422 | if err := tx.Commit(); err != nil { |
| 423 | c.mutationMu.Unlock() |
| 424 | return err |
| 425 | } |
| 426 | c.mutationMu.Unlock() |
| 427 | releaseGenerations() |
| 428 | maps.Copy(dirty, committedDirty) |
| 429 | c.publishRevision(revision, mapKeys(roots), "repair_batch") |
| 430 | c.refreshCounts(ctx) |
| 431 | return nil |
| 432 | } |
| 433 | |
| 434 | func (c *Catalog) repairBatchTestError(stage string) error { |
| 435 | if c.testRepairBatchError == nil { |
| 436 | return nil |
| 437 | } |
| 438 | return c.testRepairBatchError(stage) |
| 439 | } |
| 440 | |
| 441 | func repairDisposition(outcome repairOutcome, now time.Time) (state string, attempts int, retryAt int64, errorKind string, health Health) { |
| 442 | if outcome.err == nil { |
| 443 | switch outcome.result.Status { |
| 444 | case agent.SessionListingRepairApplied, agent.SessionListingRepairAlreadyCurrent: |
| 445 | return "complete", 0, 0, "", HealthOK |
| 446 | case agent.SessionListingRepairDamaged: |
| 447 | return "blocked", outcome.item.attempts + 1, 0, "damaged", HealthCorrupt |
| 448 | case agent.SessionListingRepairUnsupported: |
| 449 | return "blocked", outcome.item.attempts + 1, 0, "unsupported", HealthDegraded |
| 450 | case agent.SessionListingRepairSourceChanged: |
| 451 | return "deferred", outcome.item.attempts, now.Add(30 * time.Second).UnixMilli(), "source_changed", HealthDegraded |
| 452 | } |
| 453 | } |
| 454 | attempts = outcome.item.attempts + 1 |
| 455 | errorKind = "io" |
| 456 | if errors.Is(outcome.err, agent.ErrSessionListingRepairBusy) { |
| 457 | errorKind = "busy" |
| 458 | } else if errors.Is(outcome.err, context.DeadlineExceeded) { |
| 459 | errorKind = "timeout" |
| 460 | } |
| 461 | return "deferred", attempts, now.Add(repairBackoff(attempts)).UnixMilli(), errorKind, HealthDegraded |
| 462 | } |
| 463 | |
| 464 | func repairBackoff(attempts int) time.Duration { |
| 465 | if attempts < 1 { |
| 466 | attempts = 1 |
| 467 | } |
| 468 | delay := 30 * time.Second |
| 469 | for i := 1; i < attempts && delay < 30*time.Minute; i++ { |
| 470 | delay *= 2 |
| 471 | } |
| 472 | if delay > 30*time.Minute { |
| 473 | return 30 * time.Minute |
| 474 | } |
| 475 | return delay |
| 476 | } |
| 477 | |
| 478 | func (c *Catalog) repairSession(workerCtx context.Context, path string) { |
| 479 | if workerCtx.Err() != nil { |
| 480 | return |
| 481 | } |
| 482 | var item repairItem |
| 483 | var turnsState TurnsState |
| 484 | item.path = path |
| 485 | item.pathKey = c.pathKey(path) |
| 486 | if err := c.db.QueryRowContext(workerCtx, `SELECT directory,scope,workspace_root,workspace_root_key,topic_id, |
| 487 | repair_attempts,repair_state,repair_retry_at,repair_source_fingerprint,turns_state |
| 488 | FROM catalog_sessions WHERE path_key=?`, item.pathKey).Scan(&item.target.Path, &item.target.Scope, |
| 489 | &item.target.WorkspaceRoot, &item.workspaceRootKey, &item.topicID, &item.attempts, &item.state, |
| 490 | &item.retryAt, &item.sourceFingerprint, &turnsState); err != nil { |
| 491 | return |
| 492 | } |
| 493 | if turnsState != TurnsUnknown { |
| 494 | ctx, cancel := context.WithTimeout(workerCtx, 30*time.Second) |
| 495 | defer cancel() |
| 496 | _, _ = agent.RepairSessionListingProjection(ctx, path) |
| 497 | return |
| 498 | } |
| 499 | c.mutationMu.Lock() |
| 500 | tx, err := c.db.BeginTx(workerCtx, nil) |
| 501 | if err != nil { |
| 502 | c.mutationMu.Unlock() |
| 503 | return |
| 504 | } |
| 505 | claimStmt, err := tx.PrepareContext(workerCtx, repairClaimSQL) |
| 506 | if err != nil { |
| 507 | _ = tx.Rollback() |
| 508 | c.mutationMu.Unlock() |
| 509 | return |
| 510 | } |
| 511 | claimed, ok, err := claimRepairItem(workerCtx, claimStmt, item, c.opts.Now()) |
| 512 | _ = claimStmt.Close() |
| 513 | if err != nil || !ok || tx.Commit() != nil { |
| 514 | _ = tx.Rollback() |
| 515 | c.mutationMu.Unlock() |
| 516 | return |
| 517 | } |
| 518 | c.mutationMu.Unlock() |
| 519 | item = claimed |
| 520 | |
| 521 | ctx, cancel := context.WithTimeout(workerCtx, 30*time.Second) |
| 522 | defer cancel() |
| 523 | result, err := agent.RepairSessionListingProjection(ctx, path) |
| 524 | dirty := map[string]DirectoryTarget{} |
| 525 | _ = c.applyRepairBatch(workerCtx, []repairOutcome{{item: item, result: result, err: err}}, dirty) |
| 526 | for _, target := range dirty { |
| 527 | c.RequestReconcile(target) |
| 528 | } |
| 529 | } |
| 530 | |
| 531 | type knownSourceState struct { |
| 532 | preview string |
| 533 | turns int |
| 534 | turnsState TurnsState |
| 535 | health Health |
| 536 | contentFingerprint string |
| 537 | } |
| 538 | |
| 539 | // preserveKnownSourceStates prevents a directory scan backed by a legacy or |
| 540 | // transient sidecar from replacing a repaired valid/corrupt source result with |
| 541 | // unknown. The content fingerprint guard makes a changed transcript unknown |
| 542 | // again until that new generation has been parsed. |
| 543 | func (c *Catalog) preserveKnownSourceStates(ctx context.Context, directory string, records []SessionRecord) ([]SessionRecord, error) { |
| 544 | needsKnownState := false |
| 545 | for i := range records { |
| 546 | if records[i].TurnsState == TurnsUnknown { |
| 547 | needsKnownState = true |
| 548 | break |
| 549 | } |
| 550 | } |
| 551 | if !needsKnownState { |
| 552 | return records, nil |
| 553 | } |
| 554 | rows, err := c.db.QueryContext(ctx, `SELECT path,preview,turns,turns_state,health,content_fingerprint |
| 555 | FROM catalog_sessions WHERE directory_key=? AND missing_since=0 AND turns_state<>'unknown'`, c.pathKey(directory)) |
| 556 | if err != nil { |
| 557 | return nil, err |
| 558 | } |
| 559 | defer rows.Close() |
| 560 | known := make(map[string]knownSourceState) |
| 561 | for rows.Next() { |
| 562 | var path string |
| 563 | var state knownSourceState |
| 564 | if err := rows.Scan(&path, &state.preview, &state.turns, &state.turnsState, &state.health, &state.contentFingerprint); err != nil { |
| 565 | return nil, err |
| 566 | } |
| 567 | known[c.pathKey(path)] = state |
| 568 | } |
| 569 | if err := rows.Err(); err != nil { |
| 570 | return nil, err |
| 571 | } |
| 572 | for i := range records { |
| 573 | state, ok := known[c.pathKey(records[i].Path)] |
| 574 | if !ok || records[i].TurnsState != TurnsUnknown { |
| 575 | continue |
| 576 | } |
| 577 | if records[i].ContentFingerprint != state.contentFingerprint { |
| 578 | fillKnownCountHints(&records[i], state.preview, state.turns) |
| 579 | continue |
| 580 | } |
| 581 | records[i].Preview = state.preview |
| 582 | records[i].Turns = state.turns |
| 583 | records[i].TurnsState = state.turnsState |
| 584 | records[i].Health = state.health |
| 585 | } |
| 586 | return records, nil |
| 587 | } |
| 588 | |
| 589 | // fillKnownCountHints keeps a changed transcript's last certified preview and |
| 590 | // count visible while it stays unknown; repair replaces them once it lands. |
| 591 | func fillKnownCountHints(record *SessionRecord, preview string, turns int) { |
| 592 | if record.Turns != 0 || strings.TrimSpace(record.Preview) != "" { |
| 593 | return |
| 594 | } |
| 595 | record.Preview, record.Turns = preview, turns |
| 596 | } |
| 597 |