| 1 | package sessioncatalog |
| 2 | |
| 3 | import "context" |
| 4 | |
| 5 | // Progress is a catalog mutation too. In particular, it must not acquire a |
| 6 | // second SQLite write lock while metadata synchronization owns a transaction. |
| 7 | func (c *Catalog) updateDirectoryScanProgress(ctx context.Context, path string, generation int64, total int) error { |
| 8 | c.mutationMu.Lock() |
| 9 | defer c.mutationMu.Unlock() |
| 10 | if c.testScanProgressWriteHook != nil { |
| 11 | c.testScanProgressWriteHook() |
| 12 | } |
| 13 | _, err := c.db.ExecContext(ctx, `UPDATE catalog_directories SET indexed=? WHERE path_key=? AND scan_generation=?`, total, c.pathKey(path), generation) |
| 14 | if err == nil && c.opts.MetadataOnly { |
| 15 | // Unchanged metadata batches only mark presence, so progress must not |
| 16 | // depend on a redundant session/topic publication to become visible. |
| 17 | c.refreshCounts(ctx) |
| 18 | } |
| 19 | return err |
| 20 | } |
| 21 | |
| 22 | func (c *Catalog) failDirectoryScan(ctx context.Context, path string, scanErr error) { |
| 23 | c.mutationMu.Lock() |
| 24 | _, _ = c.db.ExecContext(ctx, `UPDATE catalog_directories SET state='degraded',error=? WHERE path_key=?`, scanErr.Error(), c.pathKey(path)) |
| 25 | c.mutationMu.Unlock() |
| 26 | c.statusMu.Lock() |
| 27 | c.status.State = StateDegraded |
| 28 | c.status.LastError = scanErr.Error() |
| 29 | c.statusMu.Unlock() |
| 30 | } |
| 31 |