worker

type Config struct {
	// Project mirrors the ingest worker's configured project tag; it
	// does not participate in node hashes.
	Project string

	// PollInterval is how often the dirty queue is polled.
	PollInterval time.Duration

	// Debounce is how long a session's dirty mark must be quiet before
	// it derives — ingest bursts settle into one derive.
	Debounce time.Duration

	// SweepInterval is the slow backstop cadence: enqueue every session
	// present in the raw layer, catching lost marks. A sweep also runs
	// once at startup.
	SweepInterval time.Duration

	// MaxDeriveLag bounds debounce starvation: a continuously
	// streaming session re-marks on every capture and never settles,
	// so a mark that has waited this long derives anyway. Live views
	// see bounded lag instead of waiting for a quiet gap.
	MaxDeriveLag time.Duration

	// SweepWindow bounds the backstop sweep to sessions with raw
	// activity in the last window, so a worker restart in a large org
	// re-enqueues recent sessions instead of stampeding the queue with
	// all of history (default 24h). Negative disables the bound and
	// sweeps every session ever captured — the escape hatch for a full
	// re-derive after a deriver fix.
	SweepWindow time.Duration

	// PageSize bounds one poll's batch. Sessions are still derived
	// strictly one at a time.
	PageSize int32

	// MaxPollBackoff caps the jittered exponential backoff applied
	// between polls while the queue is unreachable (DB outage). The
	// first failure retries after roughly PollInterval; each further
	// consecutive failure doubles the delay up to this cap.
	MaxPollBackoff time.Duration

	// DrainTimeout bounds the graceful-shutdown window: after the run
	// context is canceled, the in-flight derive may keep running this
	// long before its store operations are aborted. Locks release
	// either way.
	DrainTimeout time.Duration

	// Metrics optionally injects a pre-built metrics surface so the
	// hosting command can mount /metrics (and serve health probes)
	// before the store is even reachable — e.g. while --wait-for-db
	// retries. NewWorker builds a fresh one when nil.
	Metrics *Metrics
}

type Metrics struct {

	// Derives counts per-session derive attempts by outcome:
	// ok / error / locked (another worker holds the session) /
	// skipped (cleared by a peer or re-dirtied inside the debounce).
	Derives *prometheus.CounterVec

	// Requeued counts successful derives whose conditional clear
	// matched nothing — the session was re-dirtied mid-derive and
	// stays queued.
	Requeued prometheus.Counter

	NodesUpserted  prometheus.Counter
	NodesPruned    prometheus.Counter
	DeriveDuration prometheus.Histogram

	// UnknownCalls counts calls the classifier could not match to any
	// cataloged tell (derive.KindUnknown). A sustained non-zero rate
	// means the harness grew a new call shape the catalog is missing.
	UnknownCalls prometheus.Counter

	// ParseFailures counts raw rows whose request or response no
	// longer parses. Non-zero means a provider parser regressed or a
	// raw row predates a parser fix — the raw row is intact and
	// re-derivable later, but the signal deserves an alert.
	ParseFailures prometheus.Counter

	Sweeps        *prometheus.CounterVec
	SweepEnqueued prometheus.Counter

	PollErrors prometheus.Counter

	// ConsecutiveFailures mirrors the worker's in-memory outage
	// counter: non-zero means the queue is currently unreachable and
	// polls are backing off. Alert on sustained non-zero.
	ConsecutiveFailures prometheus.Gauge

	// QueueDepth and DeriveLag are refreshed once per successful poll:
	// how many sessions are dirty, and how stale the oldest dirty mark
	// is (now - oldest dirtied_at, seconds; 0 when the queue is empty).
	QueueDepth prometheus.Gauge
	DeriveLag  prometheus.Gauge
	// contains filtered or unexported fields
}

type Store interface {
	// ListDeriveDirty returns dirty sessions that have settled OR
	// waited past the max-lag bound, oldest first.
	ListDeriveDirty(ctx context.Context, dirtiedBefore, firstDirtiedBefore time.Time, limit int32) ([]storage.DeriveQueueEntry, error)

	// GetDeriveDirty re-reads one queue entry (nil when clean).
	GetDeriveDirty(ctx context.Context, orgID, harnessID, harnessSessionID string) (*storage.DeriveQueueEntry, error)

	// ClearDeriveDirty removes the entry only if DirtiedAt is unchanged.
	ClearDeriveDirty(ctx context.Context, e storage.DeriveQueueEntry) (bool, error)

	// SweepDeriveDirty enqueues every raw-layer session active at or
	// after activeSince (zero time = everything).
	SweepDeriveDirty(ctx context.Context, activeSince time.Time) (int64, error)

	// DeriveQueueStats reports queue depth and the oldest dirty mark
	// (the worker's depth/lag gauges and readiness probe).
	DeriveQueueStats(ctx context.Context) (storage.DeriveQueueStats, error)

	// TryDeriveSessionLock takes the per-session advisory lock.
	// acquired=false (nil error) means another worker holds it.
	TryDeriveSessionLock(ctx context.Context, orgID, harnessID, harnessSessionID string) (release func(), acquired bool, err error)

	// RederiveSession re-derives and persists one harness session.
	RederiveSession(ctx context.Context, project, orgID, harnessID, harnessSessionID string) (*derive.RederiveReport, error)
}

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