loop

Documentation

Overview

Package loop is the PR-71 orchestrator that connects the Discovery, Dispatch, and Render layers into one tick:

1. discover: walk every plugin in the source.Registry, call Sincer.Since when supported (else Plugin.List), and upsert results into the tasks table. A plugin's failure does not abort the rest of the iteration — other sources still run and the dispatch / render phases still fire.
2. dispatch: invoke dispatch.Dispatcher.Run with the configured MaxParallel. The dispatcher honours the existing lock_key / ClaimWorkspace concurrency contract.
3. render: invoke the Render hook so ~/.marunage/view.md reflects the post-dispatch state.

The package owns no I/O of its own beyond the discover-side store upserts: dispatch and render are passed in as narrow interfaces so the CLI wires the production implementations and tests can drop in fakes.

Concurrency: a Loop is safe for concurrent RunOnce / Run calls only when WithLockKey is set; the kv_state-backed lock is the requirement.md "lock_key を尊重した並列ループ" guarantee. Without WithLockKey, callers are responsible for serialising their own ticks.

Index

• Variables
• type Dispatcher
• type KVStateRepo
• type Loop
  • func New(opts ...Option) (*Loop, error)
  • func (l *Loop) Run(ctx context.Context, interval time.Duration) error
  • func (l *Loop) RunOnce(ctx context.Context) (err error)
• type Option
  • func WithAuditor(a config.Auditor) Option
  • func WithClock(now func() time.Time) Option
  • func WithDispatcher(d Dispatcher) Option
  • func WithKVStateRepo(r KVStateRepo) Option
  • func WithLockKey(key string) Option
  • func WithMaxParallel(n int) Option
  • func WithRegistry(r *source.Registry) Option
  • func WithRender(r Render) Option
  • func WithTaskRepo(r TaskRepo) Option
• type Render
• type TaskRepo

Variables

var ErrInvalidConfig = errors.New("loop: missing required option")

ErrInvalidConfig signals a missing required Option at construction.

var ErrInvalidInterval = errors.New("loop: interval must be > 0")

ErrInvalidInterval is returned by Run when the supplied interval is non-positive — a zero-or-negative tick would either spin the loop hot or never fire, both of which are configuration bugs the daemon should flag rather than silently absorb.

var ErrLockBusy = errors.New("loop: lock busy")

ErrLockBusy is returned by RunOnce when WithLockKey is configured and another writer (another loop process, a stuck previous tick) currently holds the lock. The CLI translates this into "skip this tick" rather than failing the daemon.

Types

type Dispatcher

type Dispatcher interface {
	Run(ctx context.Context, opts dispatch.RunOptions) error
}

Dispatcher is the dispatch surface the loop needs. *dispatch.Dispatcher satisfies it.

type KVStateRepo

type KVStateRepo interface {
	Get(ctx context.Context, key string) (string, error)
	Set(ctx context.Context, key, value string) error
	InsertIfAbsent(ctx context.Context, key, value string) (bool, error)
	DeleteIfValue(ctx context.Context, key, expected string) (bool, error)
}

KVStateRepo is the narrow surface the loop needs against the kv_state table. *store.KVStateRepo satisfies it.

InsertIfAbsent + DeleteIfValue are the atomic primitives the loop's lock acquire / release path uses. Together with an owner token they give mutual exclusion that survives a panic + a crash + a concurrent re-acquire by another process — see the matching primitives' godoc on *store.KVStateRepo for the SQL contract.

type Loop

type Loop struct {
	// contains filtered or unexported fields
}

Loop is the orchestrator. One instance per process; concurrency guarantees rest on WithLockKey.

func New

func New(opts ...Option) (*Loop, error)

New builds a Loop. Required: WithRegistry, WithTaskRepo, WithDispatcher, WithRender. Returns ErrInvalidConfig naming the missing field so a buggy CLI wiring fails loud at startup.

func (*Loop) Run

func (l *Loop) Run(ctx context.Context, interval time.Duration) error

Run drives RunOnce immediately and then on every interval tick until ctx is cancelled. RunOnce errors are audited but do not stop the loop — the next tick still runs. This mirrors the dispatch contract: per-plugin failures do not poison the queue.

func (*Loop) RunOnce

func (l *Loop) RunOnce(ctx context.Context) (err error)

RunOnce performs one full discover -> dispatch -> render iteration. Per-plugin Discovery failures are isolated: they record an audit entry and let the rest of the iteration continue. Dispatch / render errors are returned to the caller so Run can audit them and decide whether to keep ticking.

type Option

type Option func(*Loop)

Option mutates Loop construction.

func WithAuditor

func WithAuditor(a config.Auditor) Option

WithAuditor installs the audit-log sink. Defaults to config.NopAuditor so tests / CLI paths that have not wired audit.log still build.

func WithClock

func WithClock(now func() time.Time) Option

WithClock injects a deterministic clock. Defaults to time.Now in production. The clock is used for the kv_state checkpoint timestamp the loop writes after a successful per-plugin discover.

func WithDispatcher

func WithDispatcher(d Dispatcher) Option

WithDispatcher injects the dispatcher. Required.

func WithKVStateRepo

func WithKVStateRepo(r KVStateRepo) Option

WithKVStateRepo injects the kv_state repository for per-source checkpoints + the optional global lock. Optional: when nil, Sincer plugins receive an empty checkpoint string and WithLockKey is a no-op.

func WithLockKey

func WithLockKey(key string) Option

WithLockKey turns on the kv_state-backed exclusion lock. When set, RunOnce attempts to claim the key before starting; concurrent calls see ErrLockBusy. The lock is released in a deferred call so even a panic from one of the inner phases unblocks the next tick. Requires WithKVStateRepo to take effect.

func WithMaxParallel

func WithMaxParallel(n int) Option

WithMaxParallel sets the dispatcher MaxParallel passed on each tick. Defaults to 1 so a misconfigured loop does not silently turn dispatch into a no-op.

func WithRegistry

func WithRegistry(r *source.Registry) Option

WithRegistry injects the plugin registry. Required.

func WithRender

func WithRender(r Render) Option

WithRender injects the render hook. Required.

func WithTaskRepo

func WithTaskRepo(r TaskRepo) Option

WithTaskRepo injects the tasks repository (Discovery upsert target). Required.

type Render

type Render interface {
	Render(ctx context.Context) error
}

Render is the render surface the loop needs. The CLI wires a closure that calls internal/render.Render and writes ~/.marunage/view.md atomically; tests pass a fake to assert "render ran exactly once".

type TaskRepo

type TaskRepo interface {
	Insert(ctx context.Context, t store.Task) (int64, error)
	List(ctx context.Context, f store.ListFilter) ([]store.Task, error)
}

TaskRepo is the narrow write/read surface the loop needs against the tasks table. The full *store.TaskRepo satisfies it implicitly so the CLI can hand the concrete type in.