orchestra/internal/operations/observations.go

package operations

import (
	"encoding/json"
	"fmt"
	"time"

	"orchestra/internal/authz"
	"orchestra/internal/domain"
	"orchestra/internal/store"
)

// QuietTimeout bounds an incident that has no lease to bound it. A worker-level
// observation has no terminal boundary of its own, so staleness of its last
// actual occurrence is the only honest closer.
const QuietTimeout = 10 * time.Minute

// ObservationTracker turns a worker's bounded, lossy ring into durable
// incidents. It is deliberately not a copy of the ring.
//
// The rules that matter, and why:
//
//   - An incident is opened at first sight and appended immediately, so a
//     coordinator that dies mid-incident still leaves the fact that it existed.
//   - A repeat updates the aggregate and appends nothing. Run 11's 409 loop
//     repeated 301 times; appending each would have been 301 pieces of evidence
//     for one problem, and would have made every debt item eligible at once.
//   - Absence from the ring closes nothing. The ring is a bounded history, so a
//     message can vanish because it was evicted rather than because it stopped.
//   - The lease is the scope. The same signature going quiet and returning
//     inside one epoch is one incident, not two recurrences.
type ObservationTracker struct {
	Store *store.Store
	// counts is the last count this tracker saw for an open incident, so a
	// ring entry that is evicted and recreated accumulates rather than
	// restarting. Reported 34, evicted, reported 3 again means 37 occurrences,
	// not 3. In-memory: a restart loses the accumulation, never the incident.
	counts map[string]int
	// incarnations is the last incarnation seen per worker, which is what makes
	// a restart detectable at all.
	incarnations map[string]string
}

// WorkerReport is one heartbeat's worth of attributed observations. The
// coordinator attributes them, because the ring carries only messages: the
// worker's active task and that task's current lease epoch are what bind an
// incident to the work it happened during.
type WorkerReport struct {
	WorkerID     string
	Incarnation  string
	TaskID       string
	LeaseEpoch   string
	Observations []domain.WorkerObservation
	At           time.Time
}

// Ingest folds one heartbeat into the durable incidents and returns the events
// it appended. Every close it decides is one of the four boundaries; none of
// them is "the message is no longer in the ring".
func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) {
	if t == nil || t.Store == nil || r.WorkerID == "" {
		return nil, nil
	}
	if t.counts == nil {
		t.counts, t.incarnations = map[string]int{}, map[string]string{}
	}
	at := r.At
	if at.IsZero() {
		at = time.Now().UTC()
	}
	var appended []domain.Event

	// A new process cannot continue the previous one's symptom, so its
	// incidents are finalized before anything this heartbeat says is folded in.
	if previous, seen := t.incarnations[r.WorkerID]; r.Incarnation != "" && seen && previous != r.Incarnation {
		closed, err := t.closeWhere(at, domain.ObservationCloseWorkerRestart, func(inc domain.ObservationIncident) bool {
			return inc.WorkerID == r.WorkerID
		})
		appended = append(appended, closed...)
		if err != nil {
			return appended, err
		}
	}
	if r.Incarnation != "" {
		t.incarnations[r.WorkerID] = r.Incarnation
	}

	open := map[string]domain.ObservationIncident{}
	for _, inc := range t.Store.OpenObservations() {
		if inc.WorkerID == r.WorkerID {
			open[inc.Key()] = inc
		}
	}
	for _, o := range r.Observations {
		signature := domain.ObservationSignature(o.Message)
		if signature == "" {
			continue
		}
		candidate := domain.ObservationIncident{
			WorkerID: r.WorkerID, Incarnation: r.Incarnation, Signature: signature,
			TaskID: r.TaskID, LeaseEpoch: r.LeaseEpoch,
		}
		existing, isOpen := open[candidate.Key()]
		if !isOpen {
			candidate.ID = domain.NewID()
			candidate.Detail = o.Message
			candidate.FirstSeen = firstOr(o.First, at)
			candidate.LastSeen = firstOr(o.Last, at)
			e, err := t.append(domain.EventObservationIncidentOpened, candidate)
			if err != nil {
				return appended, err
			}
			appended = append(appended, e)
			t.counts[candidate.ID] = o.Count
			continue
		}
		// Open already: accumulate, append nothing. A count lower than the last
		// one means the ring evicted the entry and started it again.
		delta := o.Count - t.counts[existing.ID]
		if delta < 0 {
			delta = o.Count
		}
		t.counts[existing.ID] += delta
		if last := firstOr(o.Last, at); last.After(existing.LastSeen) {
			existing.LastSeen = last
			t.Store.NoteObservation(existing) // last_seen is durable at close
		}
	}

	// The boundaries. A lease that ended, an epoch that changed, and a
	// worker-level incident whose last occurrence has gone stale.
	closed, err := t.closeWhere(at, "", func(inc domain.ObservationIncident) bool {
		if inc.WorkerID != r.WorkerID {
			return false
		}
		if inc.TaskID == "" {
			return at.Sub(inc.LastSeen) > QuietTimeout
		}
		return inc.TaskID != r.TaskID || inc.LeaseEpoch != r.LeaseEpoch
	})
	appended = append(appended, closed...)
	return appended, err
}

// closeWhere finalizes every open incident the predicate selects. A reason of
// "" is resolved per incident, which is what lets one sweep close a lease that
// ended and a worker-level incident that went quiet.
func (t *ObservationTracker) closeWhere(at time.Time, reason domain.ObservationCloseReason, match func(domain.ObservationIncident) bool) ([]domain.Event, error) {
	var out []domain.Event
	for _, inc := range t.Store.OpenObservations() {
		if !match(inc) {
			continue
		}
		inc.ClosedAt = at
		inc.RepeatCount = t.counts[inc.ID]
		inc.CloseReason = reason
		if inc.CloseReason == "" {
			switch {
			case inc.TaskID == "":
				inc.CloseReason = domain.ObservationCloseQuietTimeout
			case inc.LeaseEpoch != "":
				inc.CloseReason = domain.ObservationCloseEpochChange
			default:
				inc.CloseReason = domain.ObservationCloseLeaseEnd
			}
		}
		e, err := t.append(domain.EventObservationIncidentClosed, inc)
		if err != nil {
			return out, err
		}
		delete(t.counts, inc.ID)
		out = append(out, e)
	}
	return out, nil
}

// append writes the incident as a worker-scoped event. The task it happened
// during is carried in the payload rather than in Event.TaskID on purpose: an
// incident is evidence about a worker, and binding it to the task aggregate
// would bump that task's version from a path the lease knows nothing about.
func (t *ObservationTracker) append(typ string, inc domain.ObservationIncident) (domain.Event, error) {
	b, err := json.Marshal(inc)
	if err != nil {
		return domain.Event{}, err
	}
	// "system" is the same aggregate QuotaReported uses for worker-scoped
	// facts: every event needs a task id, and this evidence belongs to a
	// worker rather than to any one task.
	e := domain.Event{ID: domain.NewID(), Type: typ, TaskID: "system", Payload: b, Surface: string(authz.System)}
	if err := t.Store.Append(e); err != nil {
		return domain.Event{}, fmt.Errorf("record observation incident: %w", err)
	}
	return e, nil
}

func firstOr(t, fallback time.Time) time.Time {
	if t.IsZero() {
		return fallback
	}
	return t
}