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
	// seen is the high-water mark per worker and signature, kept after an
	// incident closes. The ring is a history, so a closed incident's entry
	// keeps being reported for as long as it survives eviction; without this,
	// every heartbeat after a quiet timeout opened the same incident again and
	// manufactured recurrence out of one old failure. Live on the first real
	// run: three signatures, four incidents each, none of them a new event.
	seen map[string]watermark
	// incarnations is the last incarnation seen per worker, which is what makes
	// a restart detectable at all.
	incarnations map[string]string
}

// watermark is the last occurrence this tracker accounted for one worker's
// signature, whether or not its incident is still open.
type watermark struct {
	last  time.Time
	count int
}

// 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{}
		t.seen = map[string]watermark{}
	}
	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
		}
		// The task comes from the message, because the ring outlives the work
		// it describes. Only a failure that names no task is attributed to the
		// lease running now.
		taskID, epoch := domain.ObservationTaskID(o.Message), ""
		if taskID == "" {
			taskID, epoch = r.TaskID, r.LeaseEpoch
		} else if taskID == r.TaskID {
			epoch = r.LeaseEpoch
		}
		candidate := domain.ObservationIncident{
			WorkerID: r.WorkerID, Incarnation: r.Incarnation, Signature: signature,
			TaskID: taskID, LeaseEpoch: epoch,
		}
		mark := t.seen[r.WorkerID+"\x00"+signature]
		existing, isOpen := open[candidate.Key()]
		if !isOpen {
			// Nothing new: this is a closed incident's entry still sitting in
			// the ring. Presence is not occurrence.
			if !firstOr(o.Last, at).After(mark.last) && o.Count <= mark.count {
				continue
			}
			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
			t.seen[r.WorkerID+"\x00"+signature] = watermark{last: candidate.LastSeen, count: 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
		}
		t.seen[r.WorkerID+"\x00"+signature] = watermark{last: existing.LastSeen, count: o.Count}
	}

	// 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
		}
		// Only an incident bound to a live lease has a lease boundary to close
		// it. One read out of the ring about a task that already finished has
		// no such boundary, so it ends the way a worker-level incident does.
		if inc.LeaseEpoch == "" {
			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.LeaseEpoch == "":
				inc.CloseReason = domain.ObservationCloseQuietTimeout
			case inc.TaskID != "":
				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
}

// RecordIntervention writes down a repair the operator made by hand. It is
// deliberately an explicit act: Orchestra cannot see a worker someone
// restarted or a transaction someone deleted, and inferring "an operator
// probably intervened" from a gap in the log would put guesses into the
// evidence the ledger is built from.
func RecordIntervention(s *store.Store, surface authz.Surface, in domain.OperatorIntervention) (domain.Event, error) {
	if err := in.Validate(); err != nil {
		return domain.Event{}, err
	}
	if in.At.IsZero() {
		in.At = time.Now().UTC()
	}
	taskID := in.TaskID
	version := 0
	if taskID != "" {
		t, ok := s.Task(taskID)
		if !ok {
			return domain.Event{}, domain.ErrNotFound
		}
		version = t.Version + 1
	} else {
		// A repair with no task is still about this deployment, so it lands on
		// the same aggregate the other worker-scoped facts use.
		taskID = "system"
	}
	b, err := json.Marshal(in)
	if err != nil {
		return domain.Event{}, err
	}
	e := domain.Event{
		ID: domain.NewID(), Type: domain.EventOperatorInterventionRecorded,
		TaskID: taskID, Version: version, At: in.At, Payload: b, Surface: string(surface),
	}
	if err := s.Append(e); err != nil {
		return domain.Event{}, err
	}
	return e, nil
}