Greg Pomerantz
93a5f879f5
- Add WriteFile method to mock filesystem (non-atomic path, creates files) - Implement WriteFileAtomic in mock with temp file + rename pattern using .tmp/ directory, matching real filesystem semantics - Update WriteFileTask.Execute() to use WriteFileAtomic - Update FlushAll() to use WriteFileAtomic - Fix mock to create files on write (matching os.WriteFile behavior) - Update tests to match new semantics (create-if-not-exists)
495 lines
13 KiB
Go
495 lines
13 KiB
Go
// Package pool provides the IO worker pool for the Pad editor.
|
|
// It follows the architecture's single-owner pattern: the logic goroutine
|
|
// owns all state, workers perform IO and post results, never touching state.
|
|
package pool
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"path/filepath"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// taskCounter generates unique task IDs.
|
|
var taskCounter atomic.Int64
|
|
|
|
// Priority determines which channel a task is dispatched to.
|
|
type Priority int
|
|
|
|
const (
|
|
HighPriority Priority = iota // UI-critical, blocks user progress
|
|
LowPriority // Background work, can be delayed
|
|
)
|
|
|
|
func (p Priority) String() string {
|
|
switch p {
|
|
case HighPriority:
|
|
return "high"
|
|
case LowPriority:
|
|
return "low"
|
|
default:
|
|
return fmt.Sprintf("unknown(%d)", p)
|
|
}
|
|
}
|
|
|
|
// TaskType identifies the kind of work for result routing.
|
|
type TaskType string
|
|
|
|
const (
|
|
// Browser tasks
|
|
TypeReadDir TaskType = "read_dir"
|
|
TypeBuildIndex TaskType = "build_index"
|
|
TypeLoadIndex TaskType = "load_index"
|
|
TypeLoadPages TaskType = "load_pages"
|
|
TypeStatDir TaskType = "stat_dir"
|
|
|
|
// File tasks
|
|
TypeReadFile TaskType = "read_file"
|
|
TypeWriteFile TaskType = "write_file"
|
|
TypeStatFile TaskType = "stat_file"
|
|
|
|
// Cache tasks
|
|
TypeWriteCache TaskType = "write_cache"
|
|
TypeReadCache TaskType = "read_cache"
|
|
TypeInvalidate TaskType = "invalidate_cache"
|
|
|
|
// State persistence
|
|
TypeSaveState TaskType = "save_state"
|
|
TypeSaveUndo TaskType = "save_undo"
|
|
)
|
|
|
|
// Task represents a unit of work to be executed by a worker.
|
|
// Tasks are immutable after creation.
|
|
type Task interface {
|
|
// Execute performs the task and returns a result.
|
|
Execute() Result
|
|
|
|
// Priority returns the task priority level.
|
|
Priority() Priority
|
|
|
|
// TaskID returns a unique identifier for this task.
|
|
TaskID() string
|
|
|
|
// TaskType returns the type of task (for result routing).
|
|
TaskType() TaskType
|
|
|
|
// DirPath returns the directory this task operates on, if any.
|
|
DirPath() string
|
|
|
|
// Context returns the context for this task, used for cancellation.
|
|
Context() context.Context
|
|
|
|
// Cancel cancels this task if it's still running.
|
|
Cancel()
|
|
|
|
// Timeout returns the timeout for this task. Zero means no timeout.
|
|
Timeout() time.Duration
|
|
}
|
|
|
|
// --- Browser Tasks ---
|
|
|
|
// ReadDirTask reads directory entries from the filesystem.
|
|
type ReadDirTask struct {
|
|
taskID string
|
|
Dir string
|
|
FS FileSystem
|
|
}
|
|
|
|
func NewReadDirTask(dir string, fs FileSystem) *ReadDirTask {
|
|
return &ReadDirTask{
|
|
taskID: fmt.Sprintf("read_dir_%s_%d", filepath.Base(dir), taskCounter.Add(1)),
|
|
Dir: dir,
|
|
FS: fs,
|
|
}
|
|
}
|
|
|
|
func (t *ReadDirTask) Execute() Result {
|
|
entries, err := t.FS.ReadDir(t.Dir)
|
|
if err != nil {
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeReadDir,
|
|
Success: false,
|
|
Error: err,
|
|
}
|
|
}
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeReadDir,
|
|
Success: true,
|
|
Data: entries,
|
|
}
|
|
}
|
|
|
|
func (t *ReadDirTask) Priority() Priority { return HighPriority }
|
|
func (t *ReadDirTask) TaskID() string { return t.taskID }
|
|
func (t *ReadDirTask) TaskType() TaskType { return TypeReadDir }
|
|
func (t *ReadDirTask) DirPath() string { return t.Dir }
|
|
func (t *ReadDirTask) Context() context.Context { return context.Background() }
|
|
func (t *ReadDirTask) Cancel() {}
|
|
func (t *ReadDirTask) Timeout() time.Duration { return 5 * time.Second }
|
|
|
|
// BuildIndexTask builds a directory index and writes it to cache.
|
|
type BuildIndexTask struct {
|
|
taskID string
|
|
Dir string
|
|
FS FileSystem
|
|
}
|
|
|
|
func NewBuildIndexTask(dir string, fs FileSystem) *BuildIndexTask {
|
|
return &BuildIndexTask{
|
|
taskID: fmt.Sprintf("build_index_%s_%d", filepath.Base(dir), taskCounter.Add(1)),
|
|
Dir: dir,
|
|
FS: fs,
|
|
}
|
|
}
|
|
|
|
func (t *BuildIndexTask) Execute() Result {
|
|
// In production, this would read the directory, sort entries,
|
|
// compute letter offsets, and write to cache.
|
|
// For now, we just signal success with the directory info.
|
|
entries, err := t.FS.ReadDir(t.Dir)
|
|
if err != nil {
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeBuildIndex,
|
|
Success: false,
|
|
Error: err,
|
|
}
|
|
}
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeBuildIndex,
|
|
Success: true,
|
|
Data: entries,
|
|
}
|
|
}
|
|
|
|
func (t *BuildIndexTask) Priority() Priority { return HighPriority }
|
|
func (t *BuildIndexTask) TaskID() string { return t.taskID }
|
|
func (t *BuildIndexTask) TaskType() TaskType { return TypeBuildIndex }
|
|
func (t *BuildIndexTask) DirPath() string { return t.Dir }
|
|
func (t *BuildIndexTask) Context() context.Context { return context.Background() }
|
|
func (t *BuildIndexTask) Cancel() {}
|
|
func (t *BuildIndexTask) Timeout() time.Duration { return 5 * time.Second }
|
|
|
|
// LoadIndexTask loads a cached directory index.
|
|
type LoadIndexTask struct {
|
|
taskID string
|
|
Dir string
|
|
FS FileSystem
|
|
}
|
|
|
|
func NewLoadIndexTask(dir string, fs FileSystem) *LoadIndexTask {
|
|
return &LoadIndexTask{
|
|
taskID: fmt.Sprintf("load_index_%s_%d", filepath.Base(dir), taskCounter.Add(1)),
|
|
Dir: dir,
|
|
FS: fs,
|
|
}
|
|
}
|
|
|
|
func (t *LoadIndexTask) Execute() Result {
|
|
// In production, this would read the index from cache.
|
|
// For now, we just signal success.
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeLoadIndex,
|
|
Success: true,
|
|
Data: nil,
|
|
}
|
|
}
|
|
|
|
func (t *LoadIndexTask) Priority() Priority { return HighPriority }
|
|
func (t *LoadIndexTask) TaskID() string { return t.taskID }
|
|
func (t *LoadIndexTask) TaskType() TaskType { return TypeLoadIndex }
|
|
func (t *LoadIndexTask) DirPath() string { return t.Dir }
|
|
func (t *LoadIndexTask) Context() context.Context { return context.Background() }
|
|
func (t *LoadIndexTask) Cancel() {}
|
|
func (t *LoadIndexTask) Timeout() time.Duration { return 5 * time.Second }
|
|
|
|
// LoadPagesTask loads specific pages of directory entries from cache.
|
|
type LoadPagesTask struct {
|
|
taskID string
|
|
Dir string
|
|
PageIndices []int
|
|
FS FileSystem
|
|
}
|
|
|
|
func NewLoadPagesTask(dir string, pageIndices []int, fs FileSystem) *LoadPagesTask {
|
|
return &LoadPagesTask{
|
|
taskID: fmt.Sprintf("load_pages_%s_%d", filepath.Base(dir), taskCounter.Add(1)),
|
|
Dir: dir,
|
|
PageIndices: pageIndices,
|
|
FS: fs,
|
|
}
|
|
}
|
|
|
|
func (t *LoadPagesTask) Execute() Result {
|
|
// In production, this would read pages from cache.
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeLoadPages,
|
|
Success: true,
|
|
Data: t.PageIndices,
|
|
}
|
|
}
|
|
|
|
func (t *LoadPagesTask) Priority() Priority { return HighPriority }
|
|
func (t *LoadPagesTask) TaskID() string { return t.taskID }
|
|
func (t *LoadPagesTask) TaskType() TaskType { return TypeLoadPages }
|
|
func (t *LoadPagesTask) DirPath() string { return t.Dir }
|
|
func (t *LoadPagesTask) Context() context.Context { return context.Background() }
|
|
func (t *LoadPagesTask) Cancel() {}
|
|
func (t *LoadPagesTask) Timeout() time.Duration { return 5 * time.Second }
|
|
|
|
// StatDirTask gets directory metadata.
|
|
type StatDirTask struct {
|
|
taskID string
|
|
Dir string
|
|
FS FileSystem
|
|
}
|
|
|
|
func NewStatDirTask(dir string, fs FileSystem) *StatDirTask {
|
|
return &StatDirTask{
|
|
taskID: fmt.Sprintf("stat_dir_%s_%d", filepath.Base(dir), taskCounter.Add(1)),
|
|
Dir: dir,
|
|
FS: fs,
|
|
}
|
|
}
|
|
|
|
func (t *StatDirTask) Execute() Result {
|
|
exists := t.FS.DirExists(t.Dir)
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeStatDir,
|
|
Success: true,
|
|
Data: map[string]bool{"exists": exists},
|
|
}
|
|
}
|
|
|
|
func (t *StatDirTask) Priority() Priority { return LowPriority }
|
|
func (t *StatDirTask) TaskID() string { return t.taskID }
|
|
func (t *StatDirTask) TaskType() TaskType { return TypeStatDir }
|
|
func (t *StatDirTask) DirPath() string { return t.Dir }
|
|
func (t *StatDirTask) Context() context.Context { return context.Background() }
|
|
func (t *StatDirTask) Cancel() {}
|
|
func (t *StatDirTask) Timeout() time.Duration { return 30 * time.Second }
|
|
|
|
// --- File Tasks ---
|
|
|
|
// ReadFileTask reads file content.
|
|
type ReadFileTask struct {
|
|
taskID string
|
|
Path string
|
|
FS FileSystem
|
|
}
|
|
|
|
func NewReadFileTask(path string, fs FileSystem) *ReadFileTask {
|
|
return &ReadFileTask{
|
|
taskID: fmt.Sprintf("read_file_%s_%d", filepath.Base(path), taskCounter.Add(1)),
|
|
Path: path,
|
|
FS: fs,
|
|
}
|
|
}
|
|
|
|
func (t *ReadFileTask) Execute() Result {
|
|
content, err := t.FS.ReadFile(t.Path)
|
|
if err != nil {
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeReadFile,
|
|
Success: false,
|
|
Error: err,
|
|
}
|
|
}
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeReadFile,
|
|
Success: true,
|
|
Data: content,
|
|
}
|
|
}
|
|
|
|
func (t *ReadFileTask) Priority() Priority { return HighPriority }
|
|
func (t *ReadFileTask) TaskID() string { return t.taskID }
|
|
func (t *ReadFileTask) TaskType() TaskType { return TypeReadFile }
|
|
func (t *ReadFileTask) DirPath() string { return filepath.Dir(t.Path) }
|
|
func (t *ReadFileTask) Context() context.Context { return context.Background() }
|
|
func (t *ReadFileTask) Cancel() {}
|
|
func (t *ReadFileTask) Timeout() time.Duration { return 5 * time.Second }
|
|
|
|
// WriteFileTask writes file content (auto-save).
|
|
type WriteFileTask struct {
|
|
taskID string
|
|
Path string
|
|
FS FileSystem
|
|
Data []byte
|
|
}
|
|
|
|
func NewWriteFileTask(path string, data []byte, fs FileSystem) *WriteFileTask {
|
|
return &WriteFileTask{
|
|
taskID: fmt.Sprintf("write_file_%s_%d", filepath.Base(path), taskCounter.Add(1)),
|
|
Path: path,
|
|
FS: fs,
|
|
Data: data,
|
|
}
|
|
}
|
|
|
|
func (t *WriteFileTask) Execute() Result {
|
|
err := t.FS.WriteFileAtomic(t.Path, t.Data)
|
|
if err != nil {
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeWriteFile,
|
|
Success: false,
|
|
Error: err,
|
|
FilePath: t.Path, // Set FilePath
|
|
}
|
|
}
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeWriteFile,
|
|
Success: true,
|
|
FilePath: t.Path, // Set FilePath
|
|
}
|
|
}
|
|
|
|
func (t *WriteFileTask) Priority() Priority { return LowPriority }
|
|
func (t *WriteFileTask) TaskID() string { return t.taskID }
|
|
func (t *WriteFileTask) TaskType() TaskType { return TypeWriteFile }
|
|
func (t *WriteFileTask) DirPath() string { return filepath.Dir(t.Path) }
|
|
func (t *WriteFileTask) Context() context.Context { return context.Background() }
|
|
func (t *WriteFileTask) Cancel() {}
|
|
func (t *WriteFileTask) Timeout() time.Duration { return 30 * time.Second }
|
|
|
|
// --- Cache Tasks ---
|
|
|
|
// WriteCacheTask writes cache data.
|
|
type WriteCacheTask struct {
|
|
taskID string
|
|
Path string
|
|
FS FileSystem
|
|
Data []byte
|
|
}
|
|
|
|
func NewWriteCacheTask(path string, data []byte, fs FileSystem) *WriteCacheTask {
|
|
return &WriteCacheTask{
|
|
taskID: fmt.Sprintf("write_cache_%s_%d", filepath.Base(path), taskCounter.Add(1)),
|
|
Path: path,
|
|
FS: fs,
|
|
Data: data,
|
|
}
|
|
}
|
|
|
|
func (t *WriteCacheTask) Execute() Result {
|
|
err := t.FS.WriteFile(t.Path, t.Data)
|
|
if err != nil {
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeWriteCache,
|
|
Success: false,
|
|
Error: err,
|
|
}
|
|
}
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeWriteCache,
|
|
Success: true,
|
|
}
|
|
}
|
|
|
|
func (t *WriteCacheTask) Priority() Priority { return LowPriority }
|
|
func (t *WriteCacheTask) TaskID() string { return t.taskID }
|
|
func (t *WriteCacheTask) TaskType() TaskType { return TypeWriteCache }
|
|
func (t *WriteCacheTask) DirPath() string { return filepath.Dir(t.Path) }
|
|
func (t *WriteCacheTask) Context() context.Context { return context.Background() }
|
|
func (t *WriteCacheTask) Cancel() {}
|
|
func (t *WriteCacheTask) Timeout() time.Duration { return 30 * time.Second }
|
|
|
|
// --- State Persistence Tasks ---
|
|
|
|
// SaveStateTask persists application state.
|
|
type SaveStateTask struct {
|
|
taskID string
|
|
Path string
|
|
FS FileSystem
|
|
Data []byte
|
|
}
|
|
|
|
func NewSaveStateTask(path string, data []byte, fs FileSystem) *SaveStateTask {
|
|
return &SaveStateTask{
|
|
taskID: fmt.Sprintf("save_state_%s_%d", filepath.Base(path), taskCounter.Add(1)),
|
|
Path: path,
|
|
FS: fs,
|
|
Data: data,
|
|
}
|
|
}
|
|
|
|
func (t *SaveStateTask) Execute() Result {
|
|
err := t.FS.WriteFile(t.Path, t.Data)
|
|
if err != nil {
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeSaveState,
|
|
Success: false,
|
|
Error: err,
|
|
}
|
|
}
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeSaveState,
|
|
Success: true,
|
|
}
|
|
}
|
|
|
|
func (t *SaveStateTask) Priority() Priority { return LowPriority }
|
|
func (t *SaveStateTask) TaskID() string { return t.taskID }
|
|
func (t *SaveStateTask) TaskType() TaskType { return TypeSaveState }
|
|
func (t *SaveStateTask) DirPath() string { return filepath.Dir(t.Path) }
|
|
func (t *SaveStateTask) Context() context.Context { return context.Background() }
|
|
func (t *SaveStateTask) Cancel() {}
|
|
func (t *SaveStateTask) Timeout() time.Duration { return 30 * time.Second }
|
|
|
|
// SaveUndoTask persists undo stack.
|
|
type SaveUndoTask struct {
|
|
taskID string
|
|
Path string
|
|
FS FileSystem
|
|
Data []byte
|
|
}
|
|
|
|
func NewSaveUndoTask(path string, data []byte, fs FileSystem) *SaveUndoTask {
|
|
return &SaveUndoTask{
|
|
taskID: fmt.Sprintf("save_undo_%s_%d", filepath.Base(path), taskCounter.Add(1)),
|
|
Path: path,
|
|
FS: fs,
|
|
Data: data,
|
|
}
|
|
}
|
|
|
|
func (t *SaveUndoTask) Execute() Result {
|
|
err := t.FS.WriteFile(t.Path, t.Data)
|
|
if err != nil {
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeSaveUndo,
|
|
Success: false,
|
|
Error: err,
|
|
}
|
|
}
|
|
return Result{
|
|
TaskID: t.taskID,
|
|
TaskType: TypeSaveUndo,
|
|
Success: true,
|
|
}
|
|
}
|
|
|
|
func (t *SaveUndoTask) Priority() Priority { return LowPriority }
|
|
func (t *SaveUndoTask) TaskID() string { return t.taskID }
|
|
func (t *SaveUndoTask) TaskType() TaskType { return TypeSaveUndo }
|
|
func (t *SaveUndoTask) DirPath() string { return filepath.Dir(t.Path) }
|
|
func (t *SaveUndoTask) Context() context.Context { return context.Background() }
|
|
func (t *SaveUndoTask) Cancel() {}
|
|
func (t *SaveUndoTask) Timeout() time.Duration { return 30 * time.Second }
|