mesh-catalog/modules/messenger/cmd/messenger/sync.go

package main

// Reading the state (novox/hq issue 271): what is open now, from the controller's `conditions`
// verb, is what this holder believes — not the events it happened to be handed. Read on start, every
// ResyncEvery, and at once when an old event arrives (a replay, a recreated consumer, the holder back
// from away). Everything before the reading is in it, so no event older than it is acted on.

import (
	"sort"
	"time"
)

// SyncResult is what a reading of the state changed, for the log and the status.
type SyncResult struct {
	Open    int      `json:"open"`
	Said    []string `json:"said,omitempty"`     // urgent and never said: said now, in one message
	News    []string `json:"news,omitempty"`     // a warning raised within FreshFor and not heard: said like any other
	State   []string `json:"as_state,omitempty"` // open, older, never said: kept, not said
	Gone    []string `json:"gone,omitempty"`     // open here, not in the controller's state: ended, quietly
	Skipped []string `json:"newer_here,omitempty"`
}

// Sync takes the controller's open conditions, read by a request made at `asked`.
func (h *Holder) Sync(open []Condition, asked time.Time) SyncResult {
	h.work.Lock()
	defer h.work.Unlock()
	h.mu.Lock()
	h.init()
	now := h.Now()
	h.mu.Unlock()
	res := SyncResult{Open: len(open)}
	in := map[string]bool{}
	for _, c := range open {
		in[c.Key] = true
	}
	// What is open here and no longer there ended while this holder was not told: forgotten without
	// a message — said only where an edit notifies nobody. What this holder heard of after the
	// reading was asked for is newer than it, and stands.
	h.mu.Lock()
	var gone []*Record
	for _, r := range h.open {
		if r.Origin == "condition" && r.Cleared.IsZero() && !in[r.Key] && !r.Seen.After(asked) {
			gone = append(gone, r)
		}
	}
	h.mu.Unlock()
	sort.Slice(gone, func(i, j int) bool { return gone[i].Key < gone[j].Key })
	for _, r := range gone {
		res.Gone = append(res.Gone, r.Key)
		h.cleared(r.Key, true)
	}
	var urgent []*Record
	var fresh []*Record
	for _, c := range open {
		rec := recordOf(c)
		h.mu.Lock()
		old := h.open[c.Key]
		var r *Record
		known := old != nil
		wasQuiet := false
		switch {
		case old != nil && old.Seen.After(asked):
			h.mu.Unlock()
			res.Skipped = append(res.Skipped, c.Key)
			continue
		case old != nil:
			wasQuiet = needsNothing(old) && !old.told()
			if !old.Cleared.IsZero() {
				// Ended here, open there: back to open, quietly — the reading is not an event.
				old.Cleared = time.Time{}
				old.Count++
			}
			old.Summary, old.Subject, old.Kind, old.SilencedTill = rec.Summary, rec.Subject, rec.Kind, rec.SilencedTill
			old.takeWords(rec)
			if rec.Severity != "" {
				old.Severity = rec.Severity
			}
			if !rec.Raised.IsZero() {
				old.Raised = rec.Raised
			}
			r = old
		default:
			n := rec
			n.Count = 1
			if n.Raised.IsZero() {
				n.Raised = now
			}
			n.Sent = map[string]string{}
			n.History = true
			h.open[n.Key] = &n
			r = &n
		}
		r.Seen = now
		told := r.told()
		silenced := r.silenced(now)
		h.mu.Unlock()
		switch {
		case told || silenced || h.queued(r.Key):
		case r.Severity == Urgent:
			urgent = append(urgent, r)
		case !known && now.Sub(r.Raised) <= FreshFor:
			fresh = append(fresh, r)
		case wasQuiet && !needsNothing(r):
			// Kept quiet as needing nothing; its words now need the operator (ADR 0258).
			fresh = append(fresh, r)
		default:
			res.State = append(res.State, r.Key)
		}
		h.persist(r)
	}
	// Every urgent condition open and never said: one message, now, whatever the grace.
	if len(urgent) > 0 {
		var ready []Channel
		for _, ch := range h.channels() {
			if ch.Ready() == nil {
				ready = append(ready, ch)
			}
		}
		for _, r := range urgent {
			res.Said = append(res.Said, r.Key)
			if len(ready) == 0 {
				h.mu.Lock()
				r.Pending = "raised"
				h.mu.Unlock()
				continue
			}
			for _, ch := range ready {
				h.enqueue(ch.Name(), item{Key: r.Key, What: "raised", Severity: Urgent, At: now})
			}
		}
		for _, ch := range ready {
			h.flushChannel(ch, now, true)
		}
	}
	for _, r := range fresh {
		res.News = append(res.News, r.Key)
		h.mu.Lock()
		r.History = false
		h.mu.Unlock()
		h.deliver(r, "raised")
	}
	h.mu.Lock()
	h.syncedAt = asked
	h.syncErr = ""
	h.wantSync = false
	h.mu.Unlock()
	if len(res.Said)+len(res.News)+len(res.Gone) > 0 || len(res.State) > 0 {
		h.Logf("[messenger] read the open conditions: %d open; said %d urgent never said, %d new, kept %d as state, %d ended meanwhile",
			res.Open, len(res.Said), len(res.News), len(res.State), len(res.Gone))
	}
	return res
}

// SyncFailed notes that the state could not be read; it is asked for again.
func (h *Holder) SyncFailed(err error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.init()
	first := h.syncErr == ""
	h.syncErr = err.Error()
	if first {
		h.Logf("[messenger] cannot read the open conditions from the controller (%v): old events are kept as state, never said", err)
	}
}

// WantsSync says whether the state should be read again: never read, read too long ago, or an old
// event arrived since.
func (h *Holder) WantsSync() bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.init()
	return h.wantSync || h.syncedAt.IsZero() || h.Now().Sub(h.syncedAt) >= ResyncEvery
}