mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
refactor: the task catalog is removed
Same situation as the session catalog: nothing read it. Its only importer was the CLI command that rebuilt it, and no product path ever opened the file that command wrote. Task snapshots and event logs were already the authoritative data and are read directly. That leaves two projections, both with real readers and both doing something a directory scan cannot: history-search maintains an inverted index for BM25, usage-catalog a precomputed rollup.
This commit is contained in:
esengine committed
1 parent
404c478814
commit
8c29942890
18 files changed
+2
-1584
No files matched your search
@@ -482,14 +482,6 @@ See [Usage Catalog](./USAGE_CATALOG.md).
|
||||
|
||||
Inspect or rebuild the disposable task projection independently:
|
||||
|
||||
```sh
|
||||
reasonix doctor catalogs [--json]
|
||||
reasonix catalogs reindex tasks [--project PATH ...] [--json]
|
||||
```
|
||||
|
||||
See [Task Catalog](./TASK_CATALOG.md) for the authoritative FileStore boundary,
|
||||
cross-project routing, and rebuild behavior.
|
||||
|
||||
### Memory diagnostics and recovery
|
||||
|
||||
Bare `/memory` shows all active project/global facts without hiding same-name
|
||||
|
||||
@@ -417,16 +417,6 @@ reasonix catalogs reindex history [--dir PATH ...] [--json]
|
||||
reasonix catalogs reindex usage [--json]
|
||||
详见 [用量 Catalog](./USAGE_CATALOG.zh-CN.md)。
|
||||
|
||||
可以独立检查或重建可丢弃的 Task 投影：
|
||||
|
||||
```sh
|
||||
reasonix doctor catalogs [--json]
|
||||
reasonix catalogs reindex tasks [--project PATH ...] [--json]
|
||||
```
|
||||
|
||||
权威 FileStore 边界、跨项目路由和重建行为见
|
||||
[Task Catalog](./TASK_CATALOG.zh-CN.md)。
|
||||
|
||||
### 记忆诊断与恢复
|
||||
|
||||
直接运行 `/memory` 会显示全部 project/global active facts，不会隐藏跨 scope 的同名条目。
|
||||
|
||||
@@ -54,8 +54,7 @@ nothing else.
|
||||
The history projection is documented in
|
||||
[History Search Catalog](./HISTORY_SEARCH_CATALOG.md).
|
||||
The usage rollup projection is documented in [Usage Catalog](./USAGE_CATALOG.md).
|
||||
Task snapshots and event logs likewise remain authoritative; the rebuildable
|
||||
cross-project projection is documented in [Task Catalog](./TASK_CATALOG.md).
|
||||
Task snapshots and event logs are likewise authoritative.
|
||||
|
||||
The global user config is named `config.toml`. Project-local config files keep
|
||||
the name `reasonix.toml`. If someone says "global reasonix.toml", they usually
|
||||
|
||||
@@ -41,8 +41,7 @@ Legacy 迁移、OS home 约定目录扫描以及其他 fallback 路径都会跳
|
||||
|
||||
JSONL、event log、metadata sidecar 和 `desktop-projects.json` 是权威的会话数据，
|
||||
列出或打开一个会话不读其它任何东西。Task snapshot 和
|
||||
event log 也仍是权威数据；可重建的跨项目投影见
|
||||
[Task Catalog](./TASK_CATALOG.zh-CN.md)。
|
||||
event log 也是权威数据。
|
||||
|
||||
全局用户配置文件名是 `config.toml`。项目本地配置文件仍叫 `reasonix.toml`。
|
||||
如果有人说“全局 reasonix.toml”，通常指的是 `<Reasonix home>/config.toml`。
|
||||
|
||||
@@ -1,26 +0,0 @@
|
||||
# Task Catalog
|
||||
|
||||
Reasonix stores the cross-project task projection at
|
||||
`<cache root>/task-catalog/v1.sqlite`. Task snapshots, event JSONL, idempotency
|
||||
files, file locks, version CAS, and leases remain authoritative. SQLite is never
|
||||
used to accept stop, cancel, requeue, or open-session commands.
|
||||
|
||||
An observed `FileStore` sends non-blocking snapshot and event hints only after
|
||||
the per-task lock is released. Snapshots are indexed eagerly in bounded batches;
|
||||
event logs are indexed lazily when a task expands, an event query arrives, or a
|
||||
new event is appended. Byte-offset and sequence checkpoints resume after an
|
||||
interruption. Corrupt snapshots and event lines degrade only that task.
|
||||
|
||||
The Desktop task center pages across current session, project, or all registered
|
||||
projects. Project keys are full SHA-256 hashes of canonical saved roots and are
|
||||
resolved before every action. The control service then re-reads the authoritative
|
||||
snapshot. Current process jobs and controller state always overlay catalog data,
|
||||
and expired leases are reconciled at read time without rewriting snapshots.
|
||||
|
||||
```sh
|
||||
reasonix doctor catalogs [--json]
|
||||
reasonix catalogs reindex tasks [--project PATH ...] [--json]
|
||||
```
|
||||
|
||||
Catalog corruption or partial indexing does not disable task control. Reindexing
|
||||
replaces only the disposable projection.
|
||||
@@ -1,21 +0,0 @@
|
||||
# Task Catalog
|
||||
|
||||
Reasonix 将跨项目任务投影保存到 `<cache root>/task-catalog/v1.sqlite`。任务 snapshot、
|
||||
event JSONL、idempotency 文件、文件锁、version CAS 和 lease 继续作为权威数据。SQLite
|
||||
绝不参与 stop、cancel、requeue 或 open-session 的接受判断。
|
||||
|
||||
Observed `FileStore` 只在 per-task lock 释放后发送非阻塞 snapshot/event 提示。Snapshot
|
||||
按有界批次优先索引；event log 仅在任务展开、事件查询或追加新事件时惰性索引。byte
|
||||
offset 和 sequence checkpoint 可在强退后续传；坏 snapshot 或 event line 只降级单个任务。
|
||||
|
||||
Desktop Task Center 对当前会话、当前项目或全部已注册项目分页。Project key 是 canonical
|
||||
已保存 root 的完整 SHA-256，每次 action 都先解析 key，再由 control service 重读权威
|
||||
snapshot。当前进程的 jobs/controller 状态始终覆盖 catalog；过期 lease 只在读取时对账，
|
||||
不回写 snapshot。
|
||||
|
||||
```sh
|
||||
reasonix doctor catalogs [--json]
|
||||
reasonix catalogs reindex tasks [--project PATH ...] [--json]
|
||||
```
|
||||
|
||||
Catalog 损坏或未完成索引不会禁用任务控制；reindex 只替换可丢弃投影。
|
||||
@@ -1,49 +0,0 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"reasonix/internal/taskcatalog"
|
||||
)
|
||||
|
||||
func init() {
|
||||
registerCatalogCommand(catalogCommand{
|
||||
name: "tasks", path: taskcatalog.DefaultPath, reindex: reindexTaskCatalog,
|
||||
completionFlags: []cliCompletionFlag{
|
||||
completionFlag("--json", cliCompletionNoValue),
|
||||
completionFlag("--project", cliCompletionPathValue),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func reindexTaskCatalog(args []string) int {
|
||||
fs := flag.NewFlagSet("catalogs reindex tasks", flag.ContinueOnError)
|
||||
var projects stringListFlag
|
||||
jsonOut := fs.Bool("json", false, "print status as JSON")
|
||||
fs.Var(&projects, "project", "project root to index; repeat for multiple projects")
|
||||
if code, ok := parseCommandFlags(fs, args); !ok {
|
||||
return code
|
||||
}
|
||||
if len(projects) == 0 {
|
||||
for _, target := range defaultSessionCatalogTargets() {
|
||||
if strings.TrimSpace(target.WorkspaceRoot) != "" {
|
||||
projects = append(projects, target.WorkspaceRoot)
|
||||
}
|
||||
}
|
||||
}
|
||||
items := make([]taskcatalog.Project, 0, len(projects))
|
||||
for _, root := range projects {
|
||||
items = append(items, taskcatalog.Project{Root: root, Label: filepath.Base(root)})
|
||||
}
|
||||
status, err := taskcatalog.Rebuild(context.Background(), "", items)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
return 1
|
||||
}
|
||||
return printCatalogStatus(status, *jsonOut)
|
||||
}
|
||||
@@ -1,50 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/taskmonitor"
|
||||
)
|
||||
|
||||
func BenchmarkWarmFirstPageHundredThousandTasks(b *testing.B) {
|
||||
catalog, err := Open(context.Background(), filepath.Join(b.TempDir(), "tasks.sqlite"))
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
b.Cleanup(func() { _ = catalog.Close(context.Background()) })
|
||||
root := b.TempDir()
|
||||
project := Project{Key: ProjectKey(root), Root: root, Label: "Benchmark"}
|
||||
if _, err := catalog.db.Exec(`INSERT INTO task_projects(project_key,project_root,project_label,state) VALUES(?,?,?,'ready')`,
|
||||
project.Key, project.Root, project.Label); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
tx, err := catalog.db.Begin()
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
for i := range 100_000 {
|
||||
task := taskmonitor.TaskSnapshot{SchemaVersion: 1, TaskID: fmt.Sprintf("task-%06d", i), SessionID: "session",
|
||||
State: taskmonitor.TaskStateSucceeded, Version: 1, CreatedAt: now, UpdatedAt: now.Add(time.Duration(i) * time.Millisecond)}
|
||||
raw, _ := json.Marshal(task)
|
||||
if _, err := tx.Exec(`INSERT INTO task_snapshots(project_key,task_id,session_id,state,version,created_at,updated_at,snapshot_json)
|
||||
VALUES(?,?,?,?,?,?,?,?)`, project.Key, task.TaskID, task.SessionID, task.State, task.Version,
|
||||
task.CreatedAt.UnixMilli(), task.UpdatedAt.UnixMilli(), raw); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
b.ResetTimer()
|
||||
for range b.N {
|
||||
if _, err := catalog.ListPage(context.Background(), PageRequest{ProjectKeys: []string{project.Key}, Limit: 50}); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,710 +0,0 @@
|
||||
// Package taskcatalog maintains a disposable cross-project projection of the
|
||||
// authoritative taskmonitor FileStore.
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/config"
|
||||
"reasonix/internal/projectiondb"
|
||||
"reasonix/internal/taskmonitor"
|
||||
)
|
||||
|
||||
const (
|
||||
SchemaVersion = 1
|
||||
DefaultLimit = 50
|
||||
MaxLimit = 200
|
||||
missingGrace = 30 * time.Second
|
||||
)
|
||||
|
||||
type Status struct {
|
||||
State string `json:"state"`
|
||||
Mode projectiondb.Mode `json:"mode"`
|
||||
Path string `json:"path,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Indexed int64 `json:"indexed"`
|
||||
Total int64 `json:"total"`
|
||||
Pending int64 `json:"pending"`
|
||||
Failed int64 `json:"failed"`
|
||||
LastError string `json:"lastError,omitempty"`
|
||||
}
|
||||
|
||||
type Project struct {
|
||||
Key string `json:"projectKey"`
|
||||
Root string `json:"projectRoot"`
|
||||
Label string `json:"projectLabel"`
|
||||
}
|
||||
|
||||
type PageRequest struct {
|
||||
ProjectKeys []string
|
||||
SessionID string
|
||||
States []string
|
||||
Query string
|
||||
Cursor string
|
||||
Limit int
|
||||
}
|
||||
|
||||
type Item struct {
|
||||
ProjectKey string `json:"projectKey"`
|
||||
ProjectLabel string `json:"projectLabel"`
|
||||
Task taskmonitor.TaskSnapshot `json:"task"`
|
||||
}
|
||||
|
||||
type Page struct {
|
||||
Items []Item `json:"items"`
|
||||
NextCursor string `json:"nextCursor"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Partial bool `json:"partial"`
|
||||
StaleCursor bool `json:"staleCursor"`
|
||||
Status Status `json:"status"`
|
||||
}
|
||||
|
||||
type EventPage struct {
|
||||
Items []taskmonitor.TaskEvent `json:"items"`
|
||||
NextSequence int `json:"nextSequence"`
|
||||
Partial bool `json:"partial"`
|
||||
}
|
||||
|
||||
type cursor struct {
|
||||
Revision uint64 `json:"r"`
|
||||
Updated int64 `json:"u"`
|
||||
Project string `json:"p"`
|
||||
Task string `json:"t"`
|
||||
}
|
||||
|
||||
type request struct {
|
||||
projectRoot string
|
||||
taskID string
|
||||
events bool
|
||||
flush chan struct{}
|
||||
}
|
||||
|
||||
type Catalog struct {
|
||||
db *sql.DB
|
||||
store *taskmonitor.FileStore
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
queue chan request
|
||||
dirtyWake chan struct{}
|
||||
dirtyProjects sync.Map
|
||||
wg sync.WaitGroup
|
||||
closing atomic.Bool
|
||||
revision atomic.Uint64
|
||||
statusMu sync.RWMutex
|
||||
status Status
|
||||
projectLocks sync.Map
|
||||
reconcileMu sync.Mutex
|
||||
reconciling map[string]bool
|
||||
registered map[string]bool
|
||||
reconcileDone bool
|
||||
closeOnce sync.Once
|
||||
closeDone chan struct{}
|
||||
closeErr error
|
||||
}
|
||||
|
||||
// DefaultPath returns the disposable task projection path under CacheDir.
|
||||
// Empty when cache is unavailable so Open falls back to an in-memory projection.
|
||||
func DefaultPath() string {
|
||||
cache := strings.TrimSpace(config.CacheDir())
|
||||
if cache == "" {
|
||||
return ""
|
||||
}
|
||||
return filepath.Join(cache, "task-catalog", "v1.sqlite")
|
||||
}
|
||||
|
||||
func ProjectKey(root string) string {
|
||||
root = filepath.Clean(strings.TrimSpace(root))
|
||||
if abs, err := filepath.Abs(root); err == nil {
|
||||
root = abs
|
||||
}
|
||||
sum := sha256.Sum256([]byte(root))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
const schema = `
|
||||
CREATE TABLE task_state(id INTEGER PRIMARY KEY CHECK(id=1),revision INTEGER NOT NULL DEFAULT 0);
|
||||
INSERT INTO task_state(id,revision) VALUES(1,0);
|
||||
CREATE TABLE task_projects(project_key TEXT PRIMARY KEY,project_root TEXT UNIQUE NOT NULL,project_label TEXT NOT NULL DEFAULT '',
|
||||
signature TEXT NOT NULL DEFAULT '',scan_generation INTEGER NOT NULL DEFAULT 0,scan_cursor TEXT NOT NULL DEFAULT '',state TEXT NOT NULL DEFAULT 'pending',
|
||||
error TEXT NOT NULL DEFAULT '',indexed INTEGER NOT NULL DEFAULT 0,total INTEGER NOT NULL DEFAULT 0,completed_at INTEGER NOT NULL DEFAULT 0);
|
||||
CREATE TABLE task_snapshots(project_key TEXT NOT NULL,task_id TEXT NOT NULL,session_id TEXT NOT NULL DEFAULT '',job_id TEXT NOT NULL DEFAULT '',
|
||||
kind TEXT NOT NULL DEFAULT '',label TEXT NOT NULL DEFAULT '',state TEXT NOT NULL,runtime_state TEXT NOT NULL DEFAULT '',runtime_lease_until INTEGER NOT NULL DEFAULT 0,
|
||||
version INTEGER NOT NULL,created_at INTEGER NOT NULL,updated_at INTEGER NOT NULL,error_code TEXT NOT NULL DEFAULT '',snapshot_fingerprint TEXT NOT NULL DEFAULT '',
|
||||
snapshot_json BLOB NOT NULL,health TEXT NOT NULL DEFAULT 'ok',missing_since INTEGER NOT NULL DEFAULT 0,seen_generation INTEGER NOT NULL DEFAULT 0,
|
||||
PRIMARY KEY(project_key,task_id),FOREIGN KEY(project_key) REFERENCES task_projects(project_key) ON DELETE CASCADE);
|
||||
CREATE TABLE task_event_sources(project_key TEXT NOT NULL,task_id TEXT NOT NULL,path TEXT NOT NULL,size INTEGER NOT NULL DEFAULT 0,indexed_offset INTEGER NOT NULL DEFAULT 0,
|
||||
fingerprint TEXT NOT NULL DEFAULT '',state TEXT NOT NULL DEFAULT 'pending',error TEXT NOT NULL DEFAULT '',PRIMARY KEY(project_key,task_id));
|
||||
CREATE TABLE task_events(project_key TEXT NOT NULL,task_id TEXT NOT NULL,sequence INTEGER NOT NULL,timestamp INTEGER NOT NULL,event_type TEXT NOT NULL,
|
||||
state TEXT NOT NULL,runtime_state TEXT NOT NULL DEFAULT '',event_json BLOB NOT NULL,PRIMARY KEY(project_key,task_id,sequence));
|
||||
CREATE INDEX idx_task_project_page ON task_snapshots(project_key,updated_at DESC,task_id);
|
||||
CREATE INDEX idx_task_state_page ON task_snapshots(state,updated_at DESC,project_key,task_id);
|
||||
CREATE INDEX idx_task_session ON task_snapshots(session_id,updated_at DESC,task_id);
|
||||
CREATE INDEX idx_task_events_page ON task_events(project_key,task_id,sequence);
|
||||
`
|
||||
|
||||
func migrations() []projectiondb.Migration {
|
||||
return []projectiondb.Migration{{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error {
|
||||
_, err := tx.ExecContext(ctx, schema)
|
||||
return err
|
||||
}}}
|
||||
}
|
||||
|
||||
func Open(ctx context.Context, path string) (*Catalog, error) {
|
||||
if path == "" {
|
||||
path = DefaultPath()
|
||||
}
|
||||
inMemory := strings.TrimSpace(path) == ""
|
||||
if inMemory {
|
||||
path = ""
|
||||
}
|
||||
handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{
|
||||
Path: path, MemoryName: "task-catalog", Migrations: migrations(), InMemory: inMemory, MaxOpenConns: 4,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
workerCtx, cancel := context.WithCancel(context.Background())
|
||||
c := &Catalog{db: handle.DB, store: taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks")), ctx: workerCtx, cancel: cancel,
|
||||
queue: make(chan request, 1024), dirtyWake: make(chan struct{}, 1), reconciling: map[string]bool{}, registered: map[string]bool{}, closeDone: make(chan struct{}),
|
||||
status: Status{State: string(handle.Status.State), Mode: handle.Status.Mode, Path: handle.Status.Path, LastError: handle.Status.LastError}}
|
||||
var revision uint64
|
||||
_ = c.db.QueryRowContext(ctx, `SELECT revision FROM task_state WHERE id=1`).Scan(&revision)
|
||||
c.revision.Store(revision)
|
||||
c.refresh(ctx)
|
||||
c.wg.Add(1)
|
||||
go c.worker()
|
||||
return c, nil
|
||||
}
|
||||
|
||||
func (c *Catalog) ObservedStore() *taskmonitor.FileStore {
|
||||
return taskmonitor.NewObservedFileStore(filepath.Join(".reasonix", "tasks"), c)
|
||||
}
|
||||
|
||||
func (c *Catalog) SnapshotChanged(projectRoot, taskID string) {
|
||||
c.enqueue(request{projectRoot: projectRoot, taskID: taskID})
|
||||
}
|
||||
func (c *Catalog) EventsChanged(projectRoot, taskID string) {
|
||||
c.enqueue(request{projectRoot: projectRoot, taskID: taskID, events: true})
|
||||
}
|
||||
|
||||
func (c *Catalog) enqueue(req request) {
|
||||
if c.closing.Load() {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case c.queue <- req:
|
||||
default:
|
||||
c.dirtyProjects.Store(req.projectRoot, true)
|
||||
c.wakeDirty()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) worker() {
|
||||
defer c.wg.Done()
|
||||
ticker := time.NewTicker(time.Minute)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if root, ok := c.takeDirtyProject(); ok {
|
||||
if project, exists, err := c.projectByRoot(c.ctx, root); err == nil && exists {
|
||||
_ = c.ReconcileProject(c.ctx, project)
|
||||
} else if err == nil {
|
||||
_, _ = c.RegisterProject(c.ctx, root, filepath.Base(root))
|
||||
}
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
c.markRegisteredProjectsDirty()
|
||||
case <-c.dirtyWake:
|
||||
case req := <-c.queue:
|
||||
if req.flush != nil {
|
||||
close(req.flush)
|
||||
continue
|
||||
}
|
||||
if req.events {
|
||||
_ = c.indexEvents(c.ctx, req.projectRoot, req.taskID)
|
||||
} else {
|
||||
_ = c.indexSnapshot(c.ctx, req.projectRoot, req.taskID, 0)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) takeDirtyProject() (string, bool) {
|
||||
var root string
|
||||
c.dirtyProjects.Range(func(key, _ any) bool {
|
||||
root, _ = key.(string)
|
||||
c.dirtyProjects.Delete(key)
|
||||
return false
|
||||
})
|
||||
return root, root != ""
|
||||
}
|
||||
|
||||
func (c *Catalog) markRegisteredProjectsDirty() {
|
||||
rows, err := c.db.QueryContext(c.ctx, `SELECT project_root FROM task_projects`)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var root string
|
||||
if rows.Scan(&root) == nil {
|
||||
c.dirtyProjects.Store(root, true)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) wakeDirty() {
|
||||
select {
|
||||
case c.dirtyWake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) Flush(ctx context.Context) error {
|
||||
done := make(chan struct{})
|
||||
select {
|
||||
case c.queue <- request{flush: done}:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) RegisterProject(ctx context.Context, root, label string) (Project, error) {
|
||||
project := normalizeProject(Project{Root: root, Label: label})
|
||||
_, err := c.db.ExecContext(ctx, `INSERT INTO task_projects(project_key,project_root,project_label,state) VALUES(?,?,?,'pending')
|
||||
ON CONFLICT(project_key) DO UPDATE SET project_root=excluded.project_root,project_label=excluded.project_label`, project.Key, project.Root, project.Label)
|
||||
c.reconcileMu.Lock()
|
||||
firstRegistration := !c.registered[project.Key]
|
||||
c.registered[project.Key] = true
|
||||
c.reconcileMu.Unlock()
|
||||
if err == nil && firstRegistration {
|
||||
c.scheduleReconcile(project)
|
||||
}
|
||||
return project, err
|
||||
}
|
||||
|
||||
// RequestReconcileProject forces a background authority scan even when the
|
||||
// project was already registered by an earlier page request.
|
||||
func (c *Catalog) RequestReconcileProject(ctx context.Context, root, label string) error {
|
||||
project, err := c.RegisterProject(ctx, root, label)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.scheduleReconcile(project)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Catalog) scheduleReconcile(project Project) {
|
||||
c.reconcileMu.Lock()
|
||||
if c.reconcileDone || c.reconciling[project.Key] {
|
||||
c.reconcileMu.Unlock()
|
||||
return
|
||||
}
|
||||
c.reconciling[project.Key] = true
|
||||
c.wg.Add(1)
|
||||
c.reconcileMu.Unlock()
|
||||
go func() {
|
||||
defer func() {
|
||||
c.reconcileMu.Lock()
|
||||
delete(c.reconciling, project.Key)
|
||||
c.reconcileMu.Unlock()
|
||||
c.wg.Done()
|
||||
}()
|
||||
_ = c.ReconcileProject(c.ctx, project)
|
||||
}()
|
||||
}
|
||||
|
||||
func (c *Catalog) ReconcileProject(ctx context.Context, project Project) error {
|
||||
project = normalizeProject(project)
|
||||
unlock := c.lockProject(project.Key)
|
||||
defer unlock()
|
||||
|
||||
tasks, err := c.store.ListTasks(ctx, project.Root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var generation int64
|
||||
err = c.db.QueryRowContext(ctx, `UPDATE task_projects SET scan_generation=scan_generation+1,state='scanning',total=? WHERE project_key=? RETURNING scan_generation`, len(tasks), project.Key).Scan(&generation)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, task := range tasks {
|
||||
if err := c.upsertSnapshot(ctx, project, task, generation); err != nil {
|
||||
continue
|
||||
}
|
||||
}
|
||||
now := time.Now().UnixMilli()
|
||||
_, _ = c.db.ExecContext(ctx, `UPDATE task_snapshots SET missing_since=CASE WHEN missing_since=0 THEN ? ELSE missing_since END,health='missing'
|
||||
WHERE project_key=? AND seen_generation<>?`, now, project.Key, generation)
|
||||
cutoff := now - missingGrace.Milliseconds()
|
||||
_, _ = c.db.ExecContext(ctx, `DELETE FROM task_events WHERE project_key=? AND task_id IN (
|
||||
SELECT task_id FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?
|
||||
)`, project.Key, project.Key, generation, cutoff)
|
||||
_, _ = c.db.ExecContext(ctx, `DELETE FROM task_event_sources WHERE project_key=? AND task_id IN (
|
||||
SELECT task_id FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?
|
||||
)`, project.Key, project.Key, generation, cutoff)
|
||||
_, _ = c.db.ExecContext(ctx, `DELETE FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?`,
|
||||
project.Key, generation, cutoff)
|
||||
tx, beginErr := c.db.BeginTx(ctx, nil)
|
||||
if beginErr != nil {
|
||||
return beginErr
|
||||
}
|
||||
if _, err = tx.ExecContext(ctx, `UPDATE task_projects SET state='ready',indexed=?,completed_at=? WHERE project_key=?`, len(tasks), now, project.Key); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
revision, err := bump(ctx, tx)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
c.revision.Store(revision)
|
||||
c.refresh(context.Background())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Catalog) indexSnapshot(ctx context.Context, root, taskID string, generation int64) error {
|
||||
project, ok, err := c.projectByRoot(ctx, root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !ok {
|
||||
project, err = c.RegisterProject(ctx, root, filepath.Base(root))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
unlock := c.lockProject(project.Key)
|
||||
defer unlock()
|
||||
|
||||
task, err := c.store.GetTask(ctx, project.Root, taskID)
|
||||
if err != nil || task == nil {
|
||||
return err
|
||||
}
|
||||
return c.upsertSnapshot(ctx, project, *task, generation)
|
||||
}
|
||||
|
||||
func (c *Catalog) upsertSnapshot(ctx context.Context, project Project, task taskmonitor.TaskSnapshot, generation int64) error {
|
||||
b, err := json.Marshal(task)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hash := sha256.Sum256(b)
|
||||
lease := int64(0)
|
||||
if !task.RuntimeLeaseUntil.IsZero() {
|
||||
lease = task.RuntimeLeaseUntil.UnixMilli()
|
||||
}
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `INSERT INTO task_snapshots(project_key,task_id,session_id,job_id,state,runtime_state,runtime_lease_until,version,
|
||||
created_at,updated_at,error_code,snapshot_fingerprint,snapshot_json,health,missing_since,seen_generation) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,'ok',0,?)
|
||||
ON CONFLICT(project_key,task_id) DO UPDATE SET session_id=excluded.session_id,job_id=excluded.job_id,state=excluded.state,
|
||||
runtime_state=excluded.runtime_state,runtime_lease_until=excluded.runtime_lease_until,version=excluded.version,created_at=excluded.created_at,
|
||||
updated_at=excluded.updated_at,error_code=excluded.error_code,snapshot_fingerprint=excluded.snapshot_fingerprint,snapshot_json=excluded.snapshot_json,
|
||||
health='ok',missing_since=0,seen_generation=excluded.seen_generation`, project.Key, task.TaskID, task.SessionID, task.JobID, task.State,
|
||||
task.RuntimeState, lease, task.Version, task.CreatedAt.UnixMilli(), task.UpdatedAt.UnixMilli(), task.ErrorCode, hex.EncodeToString(hash[:]), b, generation)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
revision, err := bump(ctx, tx)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
c.revision.Store(revision)
|
||||
c.refresh(context.Background())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Catalog) indexEvents(ctx context.Context, root, taskID string) error {
|
||||
if taskID == "" || filepath.Base(taskID) != taskID || strings.ContainsAny(taskID, `/\\`) {
|
||||
return errors.New("invalid task id")
|
||||
}
|
||||
project, ok, err := c.projectByRoot(ctx, root)
|
||||
if err != nil || !ok {
|
||||
return err
|
||||
}
|
||||
unlock := c.lockProject(project.Key)
|
||||
defer unlock()
|
||||
|
||||
path := filepath.Join(project.Root, ".reasonix", "tasks", taskID, "events.jsonl")
|
||||
info, statErr := os.Stat(path)
|
||||
if errors.Is(statErr, os.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
if statErr != nil {
|
||||
return statErr
|
||||
}
|
||||
fingerprint := fmt.Sprintf("%d:%d", info.Size(), info.ModTime().UnixNano())
|
||||
var oldPath, oldFingerprint string
|
||||
var oldSize, offset int64
|
||||
err = c.db.QueryRowContext(ctx, `SELECT path,size,indexed_offset,fingerprint FROM task_event_sources WHERE project_key=? AND task_id=?`,
|
||||
project.Key, taskID).Scan(&oldPath, &oldSize, &offset, &oldFingerprint)
|
||||
reset := errors.Is(err, sql.ErrNoRows) || oldPath != path || info.Size() < oldSize || info.Size() == oldSize && fingerprint != oldFingerprint
|
||||
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||||
return err
|
||||
}
|
||||
if !reset && info.Size() == oldSize && fingerprint == oldFingerprint {
|
||||
return nil
|
||||
}
|
||||
if reset {
|
||||
offset = 0
|
||||
}
|
||||
tail, err := c.store.ReadEventTail(ctx, project.Root, taskID, offset)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
reset = reset || tail.Reset
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if reset {
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM task_events WHERE project_key=? AND task_id=?`, project.Key, taskID); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, event := range tail.Items {
|
||||
b, _ := json.Marshal(event)
|
||||
if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO task_events(project_key,task_id,sequence,timestamp,event_type,state,runtime_state,event_json)
|
||||
VALUES(?,?,?,?,?,?,?,?)`, project.Key, taskID, event.Sequence, event.Timestamp.UnixMilli(), event.EventType, event.State, event.RuntimeState, b); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `INSERT INTO task_event_sources(project_key,task_id,path,size,indexed_offset,fingerprint,state,error)
|
||||
VALUES(?,?,?,?,?,?,'ready','') ON CONFLICT(project_key,task_id) DO UPDATE SET path=excluded.path,size=excluded.size,
|
||||
indexed_offset=excluded.indexed_offset,fingerprint=excluded.fingerprint,state='ready',error=''`, project.Key, taskID, path,
|
||||
info.Size(), tail.NextOffset, fingerprint)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func (c *Catalog) ListPage(ctx context.Context, req PageRequest) (Page, error) {
|
||||
status := c.Status()
|
||||
out := Page{Items: []Item{}, Revision: c.revision.Load(), Partial: status.Pending > 0, Status: status}
|
||||
limit := req.Limit
|
||||
if limit <= 0 {
|
||||
limit = DefaultLimit
|
||||
}
|
||||
if limit > MaxLimit {
|
||||
limit = MaxLimit
|
||||
}
|
||||
cur, err := decodeCursor(req.Cursor)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
if cur != nil && cur.Revision != out.Revision {
|
||||
out.StaleCursor = true
|
||||
return out, nil
|
||||
}
|
||||
where := []string{`s.health='ok'`, `s.missing_since=0`}
|
||||
args := []any{}
|
||||
if len(req.ProjectKeys) > 0 {
|
||||
parts := make([]string, len(req.ProjectKeys))
|
||||
for i, key := range req.ProjectKeys {
|
||||
parts[i] = "?"
|
||||
args = append(args, key)
|
||||
}
|
||||
where = append(where, `s.project_key IN (`+strings.Join(parts, ",")+`)`)
|
||||
}
|
||||
if req.SessionID != "" {
|
||||
where = append(where, `s.session_id=?`)
|
||||
args = append(args, req.SessionID)
|
||||
}
|
||||
if len(req.States) > 0 {
|
||||
parts := make([]string, len(req.States))
|
||||
for i, state := range req.States {
|
||||
parts[i] = "?"
|
||||
args = append(args, state)
|
||||
}
|
||||
where = append(where, `s.state IN (`+strings.Join(parts, ",")+`)`)
|
||||
}
|
||||
if query := strings.ToLower(strings.TrimSpace(req.Query)); query != "" {
|
||||
where = append(where, `(lower(s.task_id) LIKE ? OR lower(s.session_id) LIKE ? OR lower(s.error_code) LIKE ?)`)
|
||||
like := "%" + query + "%"
|
||||
args = append(args, like, like, like)
|
||||
}
|
||||
if cur != nil {
|
||||
where = append(where, `(s.updated_at<? OR (s.updated_at=? AND s.project_key>?) OR (s.updated_at=? AND s.project_key=? AND s.task_id>?))`)
|
||||
args = append(args, cur.Updated, cur.Updated, cur.Project, cur.Updated, cur.Project, cur.Task)
|
||||
}
|
||||
args = append(args, limit+1)
|
||||
rows, err := c.db.QueryContext(ctx, `SELECT s.project_key,p.project_label,s.snapshot_json FROM task_snapshots s JOIN task_projects p ON p.project_key=s.project_key
|
||||
WHERE `+strings.Join(where, ` AND `)+` ORDER BY s.updated_at DESC,s.project_key,s.task_id LIMIT ?`, args...)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var item Item
|
||||
var raw []byte
|
||||
if err := rows.Scan(&item.ProjectKey, &item.ProjectLabel, &raw); err != nil {
|
||||
return out, err
|
||||
}
|
||||
if json.Unmarshal(raw, &item.Task) != nil {
|
||||
continue
|
||||
}
|
||||
item.Task.ReconcileRuntime(time.Now())
|
||||
out.Items = append(out.Items, item)
|
||||
}
|
||||
if len(out.Items) > limit {
|
||||
out.Items = out.Items[:limit]
|
||||
last := out.Items[len(out.Items)-1]
|
||||
out.NextCursor = encodeCursor(cursor{Revision: out.Revision, Updated: last.Task.UpdatedAt.UnixMilli(), Project: last.ProjectKey, Task: last.Task.TaskID})
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (c *Catalog) ListEventPage(ctx context.Context, projectKey, taskID string, after, limit int) (EventPage, error) {
|
||||
out := EventPage{Items: []taskmonitor.TaskEvent{}, NextSequence: after}
|
||||
project, ok, err := c.Project(ctx, projectKey)
|
||||
if err != nil || !ok {
|
||||
return out, err
|
||||
}
|
||||
if err := c.indexEvents(ctx, project.Root, taskID); err != nil {
|
||||
out.Partial = true
|
||||
return out, nil
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = DefaultLimit
|
||||
}
|
||||
if limit > MaxLimit {
|
||||
limit = MaxLimit
|
||||
}
|
||||
rows, err := c.db.QueryContext(ctx, `SELECT event_json FROM task_events WHERE project_key=? AND task_id=? AND sequence>? ORDER BY sequence LIMIT ?`, projectKey, taskID, after, limit)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var raw []byte
|
||||
var event taskmonitor.TaskEvent
|
||||
if err := rows.Scan(&raw); err != nil {
|
||||
return out, err
|
||||
}
|
||||
if json.Unmarshal(raw, &event) == nil {
|
||||
out.Items = append(out.Items, event)
|
||||
out.NextSequence = event.Sequence
|
||||
}
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (c *Catalog) Project(ctx context.Context, key string) (Project, bool, error) {
|
||||
var project Project
|
||||
err := c.db.QueryRowContext(ctx, `SELECT project_key,project_root,project_label FROM task_projects WHERE project_key=?`, key).Scan(&project.Key, &project.Root, &project.Label)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return project, false, nil
|
||||
}
|
||||
return project, err == nil, err
|
||||
}
|
||||
|
||||
func (c *Catalog) projectByRoot(ctx context.Context, root string) (Project, bool, error) {
|
||||
return c.Project(ctx, ProjectKey(root))
|
||||
}
|
||||
|
||||
func bump(ctx context.Context, tx *sql.Tx) (uint64, error) {
|
||||
if _, err := tx.ExecContext(ctx, `UPDATE task_state SET revision=revision+1 WHERE id=1`); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var revision uint64
|
||||
err := tx.QueryRowContext(ctx, `SELECT revision FROM task_state WHERE id=1`).Scan(&revision)
|
||||
return revision, err
|
||||
}
|
||||
|
||||
func encodeCursor(value cursor) string {
|
||||
b, _ := json.Marshal(value)
|
||||
return base64.RawURLEncoding.EncodeToString(b)
|
||||
}
|
||||
|
||||
func decodeCursor(value string) (*cursor, error) {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return nil, nil
|
||||
}
|
||||
b, err := base64.RawURLEncoding.DecodeString(value)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid task cursor: %w", err)
|
||||
}
|
||||
var out cursor
|
||||
if json.Unmarshal(b, &out) != nil || out.Task == "" {
|
||||
return nil, errors.New("invalid task cursor")
|
||||
}
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
func (c *Catalog) refresh(ctx context.Context) {
|
||||
var indexed, total, pending, failed int64
|
||||
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_snapshots WHERE health='ok'`).Scan(&indexed)
|
||||
_ = c.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(total),0) FROM task_projects`).Scan(&total)
|
||||
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_projects WHERE state<>'ready'`).Scan(&pending)
|
||||
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_snapshots WHERE health='corrupt'`).Scan(&failed)
|
||||
c.statusMu.Lock()
|
||||
c.status.Revision, c.status.Indexed, c.status.Total, c.status.Pending, c.status.Failed = c.revision.Load(), indexed, total, pending, failed
|
||||
c.statusMu.Unlock()
|
||||
}
|
||||
|
||||
func (c *Catalog) Status() Status {
|
||||
c.statusMu.RLock()
|
||||
defer c.statusMu.RUnlock()
|
||||
return c.status
|
||||
}
|
||||
|
||||
func (c *Catalog) Close(ctx context.Context) error {
|
||||
if c == nil {
|
||||
return nil
|
||||
}
|
||||
c.closeOnce.Do(func() {
|
||||
c.closing.Store(true)
|
||||
c.cancel()
|
||||
c.reconcileMu.Lock()
|
||||
c.reconcileDone = true
|
||||
c.reconcileMu.Unlock()
|
||||
go func() {
|
||||
c.wg.Wait()
|
||||
c.closeErr = c.db.Close()
|
||||
close(c.closeDone)
|
||||
}()
|
||||
})
|
||||
select {
|
||||
case <-c.closeDone:
|
||||
return c.closeErr
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
@@ -1,97 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/taskmonitor"
|
||||
)
|
||||
|
||||
func snapshot(id, session string, version uint64, updated time.Time) taskmonitor.TaskSnapshot {
|
||||
return taskmonitor.TaskSnapshot{SchemaVersion: 1, TaskID: id, SessionID: session, State: taskmonitor.TaskStateRunning,
|
||||
RuntimeState: taskmonitor.RuntimeStateAlive, RuntimeLeaseUntil: updated.Add(time.Hour), Version: version,
|
||||
CreatedAt: updated.Add(-time.Minute), UpdatedAt: updated}
|
||||
}
|
||||
|
||||
func TestObservedStoreIndexesSnapshotsAndEvents(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
projectRoot := t.TempDir()
|
||||
catalog, err := Open(ctx, filepath.Join(t.TempDir(), "tasks.sqlite"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
|
||||
project, err := catalog.RegisterProject(ctx, projectRoot, "Demo")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store := catalog.ObservedStore()
|
||||
now := time.Now()
|
||||
if err := store.SaveTask(ctx, projectRoot, snapshot("task-1", "session-1", 1, now)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.AppendAuditEvent(ctx, projectRoot, taskmonitor.TaskEvent{Timestamp: now, EventType: "state_change", TaskID: "task-1", SessionID: "session-1", State: taskmonitor.TaskStateRunning}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
flushCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer cancel()
|
||||
if err := catalog.Flush(flushCtx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
page, err := catalog.ListPage(ctx, PageRequest{ProjectKeys: []string{project.Key}, SessionID: "session-1", Limit: 50})
|
||||
if err != nil || len(page.Items) != 1 || page.Items[0].Task.TaskID != "task-1" {
|
||||
t.Fatalf("page=%#v err=%v", page, err)
|
||||
}
|
||||
events, err := catalog.ListEventPage(ctx, project.Key, "task-1", 0, 50)
|
||||
if err != nil || len(events.Items) != 1 || events.NextSequence != 1 {
|
||||
t.Fatalf("events=%#v err=%v", events, err)
|
||||
}
|
||||
if err := store.AppendAuditEvent(ctx, projectRoot, taskmonitor.TaskEvent{Timestamp: now.Add(time.Second), EventType: "state_change",
|
||||
TaskID: "task-1", SessionID: "session-1", State: taskmonitor.TaskStateSucceeded}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := catalog.Flush(flushCtx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
events, err = catalog.ListEventPage(ctx, project.Key, "task-1", 1, 50)
|
||||
if err != nil || len(events.Items) != 1 || events.Items[0].Sequence != 2 {
|
||||
t.Fatalf("incremental events=%#v err=%v", events, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPageCursorIsRevisionBound(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
root := t.TempDir()
|
||||
catalog, err := Open(ctx, filepath.Join(t.TempDir(), "tasks.sqlite"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
|
||||
project, _ := catalog.RegisterProject(ctx, root, "Demo")
|
||||
store := catalog.ObservedStore()
|
||||
now := time.Now()
|
||||
for i, id := range []string{"a", "b", "c"} {
|
||||
if err := store.SaveTask(ctx, root, snapshot(id, "session", 1, now.Add(time.Duration(i)*time.Minute))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
flushCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer cancel()
|
||||
_ = catalog.Flush(flushCtx)
|
||||
first, err := catalog.ListPage(ctx, PageRequest{ProjectKeys: []string{project.Key}, Limit: 2})
|
||||
if err != nil || len(first.Items) != 2 || first.NextCursor == "" {
|
||||
t.Fatalf("first=%#v err=%v", first, err)
|
||||
}
|
||||
if err := store.SaveTask(ctx, root, snapshot("d", "session", 1, now.Add(4*time.Minute))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = catalog.Flush(flushCtx)
|
||||
stale, err := catalog.ListPage(ctx, PageRequest{ProjectKeys: []string{project.Key}, Cursor: first.NextCursor})
|
||||
if err != nil || !stale.StaleCursor || len(stale.Items) != 0 {
|
||||
t.Fatalf("stale=%#v err=%v", stale, err)
|
||||
}
|
||||
}
|
||||
@@ -1,29 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestCloseCanBeAwaitedAfterCanceledCaller(t *testing.T) {
|
||||
catalog, err := Open(context.Background(), filepath.Join(t.TempDir(), "tasks.sqlite"))
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
canceled, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
if err := catalog.Close(canceled); err != nil && !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("first close: %v", err)
|
||||
}
|
||||
ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer stop()
|
||||
if err := catalog.Close(ctx); err != nil {
|
||||
t.Fatalf("await eventual close: %v", err)
|
||||
}
|
||||
if err := catalog.db.Ping(); err == nil {
|
||||
t.Fatal("database remained open after close completion")
|
||||
}
|
||||
}
|
||||
@@ -1,10 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import "sync"
|
||||
|
||||
func (c *Catalog) lockProject(projectKey string) func() {
|
||||
value, _ := c.projectLocks.LoadOrStore(projectKey, &sync.Mutex{})
|
||||
mu := value.(*sync.Mutex)
|
||||
mu.Lock()
|
||||
return mu.Unlock
|
||||
}
|
||||
@@ -1,86 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/projectiondb"
|
||||
"reasonix/internal/taskmonitor"
|
||||
)
|
||||
|
||||
// Rebuild replaces only the disposable task projection after replaying the
|
||||
// authoritative snapshots for the supplied projects into a validated sibling
|
||||
// database. Event logs remain lazily indexed when a task is expanded.
|
||||
func Rebuild(ctx context.Context, path string, projects []Project) (Status, error) {
|
||||
if path == "" {
|
||||
path = DefaultPath()
|
||||
}
|
||||
if strings.TrimSpace(path) == "" {
|
||||
catalog, err := Open(ctx, "")
|
||||
if err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
catalog.reconcileMu.Lock()
|
||||
catalog.reconcileDone = true
|
||||
catalog.reconcileMu.Unlock()
|
||||
if err := reconcileProjects(ctx, catalog, projects); err != nil {
|
||||
closeRebuildCatalog(catalog)
|
||||
return catalog.Status(), err
|
||||
}
|
||||
status := catalog.Status()
|
||||
closeRebuildCatalog(catalog)
|
||||
return status, nil
|
||||
}
|
||||
err := projectiondb.Rebuild(ctx, projectiondb.OpenOptions{
|
||||
Path: path, MemoryName: "task-catalog-rebuild", Migrations: migrations(),
|
||||
}, func(ctx context.Context, db *sql.DB) error {
|
||||
catalog := &Catalog{
|
||||
db: db, store: taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks")),
|
||||
status: Status{State: "ready", Mode: projectiondb.ModeDisk, Path: path},
|
||||
reconciling: map[string]bool{}, registered: map[string]bool{}, reconcileDone: true,
|
||||
}
|
||||
return reconcileProjects(ctx, catalog, projects)
|
||||
})
|
||||
if err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
published, err := Open(ctx, path)
|
||||
if err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
status := published.Status()
|
||||
closeRebuildCatalog(published)
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func reconcileProjects(ctx context.Context, catalog *Catalog, projects []Project) error {
|
||||
for _, input := range projects {
|
||||
project, err := catalog.RegisterProject(ctx, input.Root, input.Label)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := catalog.ReconcileProject(ctx, project); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func normalizeProject(project Project) Project {
|
||||
project.Root = filepath.Clean(strings.TrimSpace(project.Root))
|
||||
if abs, err := filepath.Abs(project.Root); err == nil {
|
||||
project.Root = abs
|
||||
}
|
||||
project.Key = ProjectKey(project.Root)
|
||||
project.Label = strings.TrimSpace(project.Label)
|
||||
return project
|
||||
}
|
||||
|
||||
func closeRebuildCatalog(catalog *Catalog) {
|
||||
closeCtx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
_ = catalog.Close(closeCtx)
|
||||
}
|
||||
@@ -1,57 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/taskmonitor"
|
||||
)
|
||||
|
||||
func TestRebuildReplacesStaleTaskProjection(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
databasePath := filepath.Join(t.TempDir(), "tasks.sqlite")
|
||||
staleRoot := t.TempDir()
|
||||
currentRoot := t.TempDir()
|
||||
store := taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks"))
|
||||
now := time.Now()
|
||||
if err := store.SaveTask(ctx, staleRoot, snapshot("stale-task", "stale-session", 1, now)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.SaveTask(ctx, currentRoot, snapshot("current-task", "current-session", 1, now)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
seed, err := Open(ctx, databasePath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
staleProject, err := seed.RegisterProject(ctx, staleRoot, "Stale")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := seed.ReconcileProject(ctx, staleProject); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := seed.Close(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
currentProject := normalizeProject(Project{Root: currentRoot, Label: "Current"})
|
||||
if _, err := Rebuild(ctx, databasePath, []Project{currentProject}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rebuilt, err := Open(ctx, databasePath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = rebuilt.Close(context.Background()) })
|
||||
stalePage, err := rebuilt.ListPage(ctx, PageRequest{ProjectKeys: []string{staleProject.Key}, Limit: 50})
|
||||
if err != nil || len(stalePage.Items) != 0 {
|
||||
t.Fatalf("stale page=%#v err=%v", stalePage, err)
|
||||
}
|
||||
currentPage, err := rebuilt.ListPage(ctx, PageRequest{ProjectKeys: []string{currentProject.Key}, Limit: 50})
|
||||
if err != nil || len(currentPage.Items) != 1 || currentPage.Items[0].Task.TaskID != "current-task" {
|
||||
t.Fatalf("current page=%#v err=%v", currentPage, err)
|
||||
}
|
||||
}
|
||||
@@ -1,236 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
|
||||
"reasonix/internal/taskmonitor"
|
||||
)
|
||||
|
||||
type sharedManager struct {
|
||||
lifecycleMu sync.Mutex
|
||||
mu sync.RWMutex
|
||||
catalog *Catalog
|
||||
pending map[string]string
|
||||
closing bool
|
||||
rebuilding bool
|
||||
generation uint64
|
||||
opening bool
|
||||
openDone chan struct{}
|
||||
openCancel context.CancelFunc
|
||||
open func(context.Context, string) (*Catalog, error)
|
||||
rebuild func(context.Context, string, []Project) (Status, error)
|
||||
}
|
||||
|
||||
var shared sharedManager
|
||||
|
||||
func ensureShared() {
|
||||
shared.start()
|
||||
}
|
||||
|
||||
func (m *sharedManager) start() {
|
||||
m.mu.Lock()
|
||||
if m.catalog != nil || m.opening || m.closing || m.rebuilding {
|
||||
m.mu.Unlock()
|
||||
return
|
||||
}
|
||||
m.generation++
|
||||
generation := m.generation
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
m.opening, m.openDone, m.openCancel = true, done, cancel
|
||||
openCatalog := m.open
|
||||
if openCatalog == nil {
|
||||
openCatalog = Open
|
||||
}
|
||||
m.mu.Unlock()
|
||||
go m.openGeneration(ctx, generation, done, openCatalog)
|
||||
}
|
||||
|
||||
func (m *sharedManager) openGeneration(ctx context.Context, generation uint64, done chan struct{}, openCatalog func(context.Context, string) (*Catalog, error)) {
|
||||
catalog, err := openCatalog(ctx, "")
|
||||
seen := map[string]bool{}
|
||||
for err == nil {
|
||||
m.mu.Lock()
|
||||
if m.closing || m.rebuilding || generation != m.generation || ctx.Err() != nil {
|
||||
m.mu.Unlock()
|
||||
_ = catalog.Close(context.Background())
|
||||
catalog = nil
|
||||
break
|
||||
}
|
||||
pending := map[string]string{}
|
||||
for root, label := range m.pending {
|
||||
if !seen[root] {
|
||||
seen[root] = true
|
||||
pending[root] = label
|
||||
}
|
||||
}
|
||||
if len(pending) == 0 {
|
||||
m.catalog = catalog
|
||||
m.pending = nil
|
||||
m.mu.Unlock()
|
||||
break
|
||||
}
|
||||
m.mu.Unlock()
|
||||
for root, label := range pending {
|
||||
_, _ = catalog.RegisterProject(ctx, root, label)
|
||||
}
|
||||
}
|
||||
m.mu.Lock()
|
||||
if m.openDone == done {
|
||||
m.opening = false
|
||||
m.openDone = nil
|
||||
m.openCancel = nil
|
||||
}
|
||||
close(done)
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
func Shared() *Catalog {
|
||||
ensureShared()
|
||||
shared.mu.RLock()
|
||||
defer shared.mu.RUnlock()
|
||||
return shared.catalog
|
||||
}
|
||||
|
||||
// ShutdownShared drains accepted notifications and cancels every shared task
|
||||
// projection worker. It is only used during process shutdown; authoritative
|
||||
// task snapshots and event logs have already committed before notifications.
|
||||
func ShutdownShared(ctx context.Context) error {
|
||||
return shared.close(ctx)
|
||||
}
|
||||
|
||||
func (m *sharedManager) close(ctx context.Context) error {
|
||||
m.lifecycleMu.Lock()
|
||||
defer m.lifecycleMu.Unlock()
|
||||
return m.closeLocked(ctx, false)
|
||||
}
|
||||
|
||||
func (m *sharedManager) closeLocked(ctx context.Context, rebuild bool) error {
|
||||
m.mu.Lock()
|
||||
m.closing = true
|
||||
m.rebuilding = rebuild
|
||||
m.generation++
|
||||
if m.openCancel != nil {
|
||||
m.openCancel()
|
||||
}
|
||||
done := m.openDone
|
||||
catalog := m.catalog
|
||||
m.catalog = nil
|
||||
if !rebuild {
|
||||
m.pending = nil
|
||||
}
|
||||
m.mu.Unlock()
|
||||
var flushErr, closeErr error
|
||||
if catalog != nil {
|
||||
flushErr = catalog.Flush(ctx)
|
||||
closeErr = catalog.Close(ctx)
|
||||
}
|
||||
if done != nil {
|
||||
select {
|
||||
case <-done:
|
||||
case <-ctx.Done():
|
||||
if closeErr == nil {
|
||||
closeErr = ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
m.mu.Lock()
|
||||
m.closing = false
|
||||
m.rebuilding = rebuild
|
||||
m.mu.Unlock()
|
||||
if flushErr != nil {
|
||||
return flushErr
|
||||
}
|
||||
return closeErr
|
||||
}
|
||||
|
||||
func RegisterSharedProject(root, label string) string {
|
||||
return shared.registerProject(root, label)
|
||||
}
|
||||
|
||||
func (m *sharedManager) registerProject(root, label string) string {
|
||||
m.start()
|
||||
key := ProjectKey(root)
|
||||
m.mu.Lock()
|
||||
if m.rebuilding {
|
||||
if m.pending == nil {
|
||||
m.pending = map[string]string{}
|
||||
}
|
||||
m.pending[root] = label
|
||||
m.mu.Unlock()
|
||||
return key
|
||||
}
|
||||
if m.closing {
|
||||
m.mu.Unlock()
|
||||
return key
|
||||
}
|
||||
if m.catalog == nil {
|
||||
if m.pending == nil {
|
||||
m.pending = map[string]string{}
|
||||
}
|
||||
m.pending[root] = label
|
||||
m.mu.Unlock()
|
||||
return key
|
||||
}
|
||||
catalog := m.catalog
|
||||
m.mu.Unlock()
|
||||
_, _ = catalog.RegisterProject(context.Background(), root, label)
|
||||
return key
|
||||
}
|
||||
|
||||
type sharedSink struct{}
|
||||
|
||||
func (sharedSink) SnapshotChanged(projectRoot, taskID string) {
|
||||
catalog := sharedCatalogForNotification(projectRoot)
|
||||
if catalog != nil {
|
||||
catalog.SnapshotChanged(projectRoot, taskID)
|
||||
}
|
||||
}
|
||||
|
||||
func (sharedSink) EventsChanged(projectRoot, taskID string) {
|
||||
catalog := sharedCatalogForNotification(projectRoot)
|
||||
if catalog != nil {
|
||||
catalog.EventsChanged(projectRoot, taskID)
|
||||
}
|
||||
}
|
||||
|
||||
// sharedCatalogForNotification is deliberately SQLite-free. ProjectionSink is
|
||||
// called after the authoritative task file lock is released, but task saves
|
||||
// still must never wait for catalog I/O. A notification received while the
|
||||
// catalog is opening is recovered by the pending project's initial reconcile.
|
||||
func sharedCatalogForNotification(projectRoot string) *Catalog {
|
||||
return shared.catalogForNotification(projectRoot)
|
||||
}
|
||||
|
||||
func (m *sharedManager) catalogForNotification(projectRoot string) *Catalog {
|
||||
m.start()
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.rebuilding {
|
||||
if m.pending == nil {
|
||||
m.pending = map[string]string{}
|
||||
}
|
||||
m.pending[projectRoot] = filepath.Base(projectRoot)
|
||||
return nil
|
||||
}
|
||||
if m.closing {
|
||||
return nil
|
||||
}
|
||||
if m.catalog == nil {
|
||||
if m.pending == nil {
|
||||
m.pending = map[string]string{}
|
||||
}
|
||||
m.pending[projectRoot] = filepath.Base(projectRoot)
|
||||
return nil
|
||||
}
|
||||
return m.catalog
|
||||
}
|
||||
|
||||
// ObservedStore remains an authoritative FileStore; only its post-commit sink
|
||||
// is shared with the disposable catalog.
|
||||
func ObservedStore() taskmonitor.WriteStore {
|
||||
ensureShared()
|
||||
return taskmonitor.NewObservedFileStore(filepath.Join(".reasonix", "tasks"), sharedSink{})
|
||||
}
|
||||
@@ -1,65 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestSharedManagerCloseFencesOpenAndAllowsRestart(t *testing.T) {
|
||||
manager := &sharedManager{}
|
||||
started := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
databasePath := filepath.Join(t.TempDir(), "tasks.sqlite")
|
||||
manager.open = func(ctx context.Context, path string) (*Catalog, error) {
|
||||
close(started)
|
||||
<-release
|
||||
return Open(ctx, databasePath)
|
||||
}
|
||||
manager.start()
|
||||
<-started
|
||||
|
||||
closed := make(chan error, 1)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
go func() { closed <- manager.close(ctx) }()
|
||||
select {
|
||||
case err := <-closed:
|
||||
t.Fatalf("close returned before in-flight open exited: %v", err)
|
||||
default:
|
||||
}
|
||||
close(release)
|
||||
if err := <-closed; err != nil {
|
||||
t.Fatalf("close after open fence: %v", err)
|
||||
}
|
||||
manager.mu.RLock()
|
||||
stale := manager.catalog
|
||||
manager.mu.RUnlock()
|
||||
if stale != nil {
|
||||
t.Fatal("stale open published a task catalog after close")
|
||||
}
|
||||
|
||||
manager.mu.Lock()
|
||||
manager.open = func(ctx context.Context, path string) (*Catalog, error) {
|
||||
return Open(ctx, databasePath)
|
||||
}
|
||||
manager.mu.Unlock()
|
||||
manager.start()
|
||||
manager.mu.RLock()
|
||||
reopened := manager.openDone
|
||||
manager.mu.RUnlock()
|
||||
if reopened == nil {
|
||||
t.Fatal("manager did not start a new generation after close")
|
||||
}
|
||||
<-reopened
|
||||
manager.mu.RLock()
|
||||
restarted := manager.catalog
|
||||
manager.mu.RUnlock()
|
||||
if restarted == nil {
|
||||
t.Fatal("manager did not publish the restarted task catalog")
|
||||
}
|
||||
if err := manager.close(ctx); err != nil {
|
||||
t.Fatalf("close restarted task catalog: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import "context"
|
||||
|
||||
// RebuildSharedCatalog fences the process-local reader, atomically replaces
|
||||
// the disposable database, and reopens it with every project registered during
|
||||
// or before the rebuild.
|
||||
func RebuildSharedCatalog(ctx context.Context, projects []Project) error {
|
||||
return shared.rebuildShared(ctx, projects)
|
||||
}
|
||||
|
||||
func (m *sharedManager) rebuildShared(ctx context.Context, projects []Project) error {
|
||||
m.lifecycleMu.Lock()
|
||||
defer m.lifecycleMu.Unlock()
|
||||
if err := m.closeLocked(ctx, true); err != nil {
|
||||
m.mu.Lock()
|
||||
m.rebuilding = false
|
||||
if m.pending == nil {
|
||||
m.pending = map[string]string{}
|
||||
}
|
||||
for _, project := range projects {
|
||||
project = normalizeProject(project)
|
||||
m.pending[project.Root] = project.Label
|
||||
}
|
||||
m.mu.Unlock()
|
||||
if ctx.Err() == nil {
|
||||
m.start()
|
||||
}
|
||||
return err
|
||||
}
|
||||
m.mu.Lock()
|
||||
if m.pending == nil {
|
||||
m.pending = map[string]string{}
|
||||
}
|
||||
for _, project := range projects {
|
||||
project = normalizeProject(project)
|
||||
m.pending[project.Root] = project.Label
|
||||
}
|
||||
rebuild := m.rebuild
|
||||
if rebuild == nil {
|
||||
rebuild = Rebuild
|
||||
}
|
||||
m.mu.Unlock()
|
||||
_, rebuildErr := rebuild(ctx, "", projects)
|
||||
m.mu.Lock()
|
||||
m.rebuilding = false
|
||||
m.mu.Unlock()
|
||||
if ctx.Err() == nil {
|
||||
m.start()
|
||||
}
|
||||
return rebuildErr
|
||||
}
|
||||
@@ -1,74 +0,0 @@
|
||||
package taskcatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestSharedManagerRebuildRestartsWithRegisteredProjects(t *testing.T) {
|
||||
manager := &sharedManager{}
|
||||
databasePath := filepath.Join(t.TempDir(), "tasks.sqlite")
|
||||
manager.open = func(ctx context.Context, _ string) (*Catalog, error) {
|
||||
return Open(ctx, databasePath)
|
||||
}
|
||||
rebuildCalls := 0
|
||||
manager.rebuild = func(context.Context, string, []Project) (Status, error) {
|
||||
rebuildCalls++
|
||||
return Status{}, nil
|
||||
}
|
||||
manager.start()
|
||||
waitSharedManagerOpen(t, manager)
|
||||
project := Project{Root: t.TempDir(), Label: "Current"}
|
||||
if err := manager.rebuildShared(context.Background(), []Project{project}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitSharedManagerOpen(t, manager)
|
||||
if rebuildCalls != 1 {
|
||||
t.Fatalf("rebuild calls = %d, want 1", rebuildCalls)
|
||||
}
|
||||
manager.mu.RLock()
|
||||
restarted := manager.catalog
|
||||
manager.mu.RUnlock()
|
||||
if restarted == nil {
|
||||
t.Fatal("manager did not publish a fresh catalog after rebuild")
|
||||
}
|
||||
normalized := normalizeProject(project)
|
||||
if _, ok, err := restarted.Project(context.Background(), normalized.Key); err != nil || !ok {
|
||||
t.Fatalf("registered project missing after rebuild: ok=%v err=%v", ok, err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := manager.close(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSharedManagerRetainsRegistrationsWhileClosingForRebuild(t *testing.T) {
|
||||
manager := &sharedManager{closing: true, rebuilding: true}
|
||||
registeredRoot := t.TempDir()
|
||||
notifiedRoot := t.TempDir()
|
||||
manager.registerProject(registeredRoot, "Registered")
|
||||
if catalog := manager.catalogForNotification(notifiedRoot); catalog != nil {
|
||||
t.Fatal("notification unexpectedly reached a catalog during rebuild")
|
||||
}
|
||||
manager.mu.RLock()
|
||||
defer manager.mu.RUnlock()
|
||||
if manager.pending[registeredRoot] != "Registered" {
|
||||
t.Fatalf("registered project was not retained: %#v", manager.pending)
|
||||
}
|
||||
if manager.pending[notifiedRoot] != filepath.Base(notifiedRoot) {
|
||||
t.Fatalf("notified project was not retained: %#v", manager.pending)
|
||||
}
|
||||
}
|
||||
|
||||
func waitSharedManagerOpen(t *testing.T, manager *sharedManager) {
|
||||
t.Helper()
|
||||
manager.mu.RLock()
|
||||
done := manager.openDone
|
||||
manager.mu.RUnlock()
|
||||
if done != nil {
|
||||
<-done
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user