poller

Documentation

Overview

Package poller drives event ingress by polling the provider's API instead of receiving webhooks. See ADR-0031 for the rationale.

"Never polls" (the everflow brand) is about LLM tokens, not provider API calls. Polling glab/gh costs zero tokens — the latency penalty (seconds vs minutes) is acceptable for refactor sweeps that run over hours/days.

Index

• type ActiveRun
• type EventDispatcher
• type Loop
  • func (l *Loop) Run(ctx context.Context)
• type RunSource
• type SaveSnapshot
• type StoreSource
  • func (s *StoreSource) ActiveRuns(ctx context.Context) ([]ActiveRun, error)

Types

type ActiveRun

type ActiveRun struct {
	RunID     string
	ForeignID string
	Provider  string
	ProjectID string
	Author    provider.User // for IsAuthor classification when synthesising events
	InFlight  map[string]provider.MR
	// LastSeenNoteIDs maps MR IID → highest note ID we've already
	// processed, across all comment streams merged (the pre-ADR-0041
	// scalar). Read at poll-start; the SaveSnapshot callback persists
	// updates. Used as provider.NoteCursor.Legacy — the floor for any
	// stream not yet present in LastSeenNoteCursors.
	LastSeenNoteIDs map[int]int64
	// LastSeenNoteCursors maps MR IID → per-stream high-water mark (see
	// provider.NoteCursor and AgentState.LastSeenNoteIDsByStream).
	LastSeenNoteCursors map[int]map[string]int64
	LastMRStates        map[int]string
}

ActiveRun is a snapshot of a Run the poller needs to inspect.

type EventDispatcher

type EventDispatcher func(ctx context.Context, runID string, event provider.Event) error

EventDispatcher is the function shape main.go's webhook dispatcher also satisfies — synthesised events flow through the same path.

type Loop

type Loop struct {
	Interval     time.Duration
	Providers    map[string]provider.Provider
	Source       RunSource
	Dispatcher   EventDispatcher
	SaveSnapshot SaveSnapshot
	Logger       *slog.Logger

	// Concurrency bounds how many Runs' pollRun calls execute at once.
	// Found live: pollOnce used to walk every active Run sequentially in
	// one goroutine — a single Run whose event triggers a real runner
	// invocation (rn.Run, which can take 15-20+ minutes) blocked the
	// entire poller from even checking any other Run for new activity
	// until that call returned, regardless of which repo either Run
	// targeted (ADR-0093). Defaults to defaultPollConcurrency if unset.
	Concurrency int
	// contains filtered or unexported fields
}

Loop runs in a goroutine. It ticks every interval, walks active Runs, queries the provider for changes since the last snapshot, and synthesises provider.Event values that it dispatches via the same path webhooks use.

Returns when ctx is cancelled.

func (*Loop) Run

func (l *Loop) Run(ctx context.Context)

type RunSource

type RunSource interface {
	ActiveRuns(ctx context.Context) ([]ActiveRun, error)
}

RunSource enumerates active Runs for the poller to walk each tick. Implementations typically wrap the workflow.RecordStore.

type SaveSnapshot

type SaveSnapshot func(ctx context.Context, runID string, noteIDs map[int]int64, noteCursors map[int]map[string]int64, mrStates map[int]string) error

SaveSnapshot is called after each successful poll for a Run to persist the updated LastSeenNoteIDs, LastSeenNoteCursors, and LastMRStates on AgentState. Typically triggers a workflow.Callback no-op transition so the values flush to the durable store.

type StoreSource

type StoreSource struct {
	Store        workflow.RecordStore
	WorkflowName string
	Decode       func([]byte) (ActiveRun, bool) // domain-specific Object unmarshaller
}

StoreSource implements RunSource against a workflow.RecordStore. Reads active Runs (RunState != finished AND AgentStatus active) and unmarshals AgentState to extract poll state.

func (*StoreSource) ActiveRuns

func (s *StoreSource) ActiveRuns(ctx context.Context) ([]ActiveRun, error)