返回 DeepSeek-Reasonix
index_queue.go
根目录 / internal / sessioncatalog / index_queue.go
1 package sessioncatalog
2
3 import (
4 "context"
5 "path/filepath"
6 "time"
7 )
8
9 // RequestIndexSession coalesces an authoritative session write by path. It is
10 // intentionally non-blocking so a saturated projection can never delay JSONL
11 // or sidecar persistence.
12 func (c *Catalog) RequestIndexSession(target DirectoryTarget, path string) bool {
13 if c == nil {
14 return false
15 }
16 path = cleanCatalogAccessPath(path)
17 if path == "" {
18 return false
19 }
20 target.Path = cleanCatalogAccessPath(target.Path)
21 if target.Path == "" {
22 target.Path = filepath.Dir(path)
23 }
24 key := queuePathKey(path)
25 request := sessionPathRequest{target: target, path: path, queueKey: key, sequence: c.mutationSeq.Add(1)}
26 c.pathQueueMu.Lock()
27 if _, loaded := c.pathQueued.Load(key); loaded {
28 c.pathQueued.Store(key, request)
29 c.pathQueueMu.Unlock()
30 return true
31 }
32 c.pathQueued.Store(key, request)
33 select {
34 case c.pathCh <- request:
35 c.pathQueueMu.Unlock()
36 return true
37 case <-c.stop:
38 c.pathQueued.Delete(key)
39 c.pathQueueMu.Unlock()
40 return false
41 default:
42 c.pathQueued.Delete(key)
43 c.pathQueueMu.Unlock()
44 return false
45 }
46 }
47
48 func (c *Catalog) sessionPathLoop() {
49 defer c.workers.Done()
50 for {
51 select {
52 case token := <-c.pathCh:
53 c.pathQueueMu.Lock()
54 queued, ok := c.pathQueued.LoadAndDelete(token.queueKey)
55 c.pathQueueMu.Unlock()
56 if !ok {
57 continue
58 }
59 request := queued.(sessionPathRequest)
60 ctx, cancel := context.WithTimeout(c.workerCtx, 30*time.Second)
61 _ = c.indexSessionPath(ctx, request.target, request.path, request.sequence)
62 cancel()
63 case <-c.stop:
64 return
65 }
66 }
67 }
68
68 lines GO