Retracted: Retract v0.1.0 through v0.1.3: v0.1.0–v0.1.2 rewrote the journal
(async RunStep + dropped seq). v0.1.3 is the last release before
typed step input and WithStepVersion. Start from v0.1.4.
Documentation
¶
Overview ¶
Package journal provides a filesystem-backed, journal-per-task implementation of durable.Store. Each task gets its own directory containing an atomic meta.json file and an append-only journal.log that records step checkpoints.
Single-process constraint ¶
JournalStore is designed for use within a single OS process. Opening the same root directory from two processes simultaneously will corrupt journal.log (concurrent appends produce interleaved bytes). The package-level registry prevents two JournalStore instances from opening the same root within one process.
Crash safety ¶
Each entry written to journal.log is framed:
[ 4 bytes: uint32 payload length ][ N bytes: JSON ][ 4 bytes: CRC32 checksum ]
On replay, entries with a mismatched checksum or a truncated tail (from a kill -9 or power loss mid-write) are detected and discarded. The preceding fully-written entries are replayed normally.
TaskInfo is written to meta.json.tmp and then renamed into place. os.Rename is atomic on POSIX systems, so a crash mid-write always leaves either the old or the new meta.json intact — never a partial file.
Index ¶
- type JournalOption
- type JournalStore
- func (s *JournalStore) Close() error
- func (s *JournalStore) DeleteTask(ctx context.Context, taskID string) error
- func (s *JournalStore) GetTask(ctx context.Context, taskID string) (durable.TaskInfo, bool, error)
- func (s *JournalStore) ListStepIDs(ctx context.Context, taskID string) ([]string, error)
- func (s *JournalStore) ListTasks(ctx context.Context) ([]durable.TaskInfo, error)
- func (s *JournalStore) LoadStep(ctx context.Context, taskID, stepID string) (durable.StepRecord, bool, error)
- func (s *JournalStore) LoadSteps(ctx context.Context, taskID string) ([]durable.StepRecord, error)
- func (s *JournalStore) PurgeTasks(ctx context.Context, status durable.TaskStatus, before time.Time) (int64, error)
- func (s *JournalStore) SaveStep(ctx context.Context, taskID string, step durable.StepRecord) error
- func (s *JournalStore) SaveTask(ctx context.Context, task durable.TaskInfo) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type JournalOption ¶
type JournalOption func(*JournalStore)
JournalOption is a functional option for configuring a JournalStore.
func WithLogger ¶
func WithLogger(logger *slog.Logger) JournalOption
WithLogger sets an slog.Logger on the JournalStore. All store operations emit structured log events through this logger at the appropriate level. If not set, a no-op logger is used and nothing is emitted.
type JournalStore ¶
type JournalStore struct {
// contains filtered or unexported fields
}
JournalStore is a durable.Store implementation backed by the local filesystem. Each task is stored in its own subdirectory under root/tasks/<task-id>/.
JournalStore is safe for concurrent use by multiple goroutines within a single OS process. It must not be shared across processes (see package doc).
Always call Close when the store is no longer needed.
func NewJournalStore ¶
func NewJournalStore(root string, opts ...JournalOption) (*JournalStore, error)
NewJournalStore opens (or creates) a journal store rooted at root. root is created if it does not exist. Returns an error if root is already open in this process — share the existing instance instead.
func (*JournalStore) Close ¶
func (s *JournalStore) Close() error
Close releases the store from the process-level registry. After Close, NewJournalStore may be called again for the same root. Always defer Close to ensure the registry entry is cleaned up.
func (*JournalStore) DeleteTask ¶
func (s *JournalStore) DeleteTask(ctx context.Context, taskID string) error
DeleteTask implements durable.Store. Removes the task directory and all its contents (meta.json + journal.log). No-op if the task does not exist.
func (*JournalStore) ListStepIDs ¶
ListStepIDs implements durable.Store.
func (*JournalStore) ListTasks ¶
ListTasks implements durable.Store. Walks the tasks directory and reads each meta.json. Results are sorted by CreatedAt descending.
func (*JournalStore) LoadStep ¶
func (s *JournalStore) LoadStep(ctx context.Context, taskID, stepID string) (durable.StepRecord, bool, error)
LoadStep implements durable.Store.
func (*JournalStore) LoadSteps ¶
func (s *JournalStore) LoadSteps(ctx context.Context, taskID string) ([]durable.StepRecord, error)
LoadSteps implements durable.Store. Reads journal.log and replays all valid framed entries. Partial or corrupt tail entries (from kill -9) are silently discarded. The last written entry per stepID is returned (upsert semantics). Results are ordered by Seq ascending.
func (*JournalStore) PurgeTasks ¶
func (s *JournalStore) PurgeTasks(ctx context.Context, status durable.TaskStatus, before time.Time) (int64, error)
PurgeTasks implements durable.Store. Deletes task directories whose status matches and whose UpdatedAt is strictly before the cutoff time.
func (*JournalStore) SaveStep ¶
func (s *JournalStore) SaveStep(ctx context.Context, taskID string, step durable.StepRecord) error
SaveStep implements durable.Store. Appends a framed JSON entry to journal.log and calls Sync to flush to disk before returning.
Upsert semantics are achieved by appending — LoadSteps returns the last written entry per stepID (last-write-wins).
Source Files