package views import ( "cmp" "context" "encoding/json" "fmt" "maps" "slices" "sync" "time" "github.com/nats-io/nats.go/jetstream" ) // Follower is a reader's picture of one state of an owner (ADR 0615 §4 to §6): it asks the owner's first-view tool at // its start, then keeps the picture from the owner's events alone, asking the tool once more only on a gap in revisions // (said in its log), never on a timer. Started while the owner is down, it keeps asking, never reading the owner's state // instead, and past Bound raises its condition naming the owner and the tool; the answer clears it. type Follower struct { Owner string // Tool is the owner's tool it asks, as the reader's invokes names it (.). Tool string State string // Ask calls the tool for the page after after. Ask func(ctx context.Context, after string) (Answer, error) // Raise and Clear are the reader's condition of this first view (past Bound; cleared by the answer). Nil says // nothing. Raise func(Status) Clear func(Status) // Changed is told of each key the picture changes (deleted: the key is gone), after the change is held, with when it // was written: the zero time for a key a first view brought (since before what the reader heard). Changed func(key string, value json.RawMessage, deleted bool, at time.Time) // Said is told of the status each time an ask starts, is answered or fails, and of a raise and a clear: where a // reader keeps it on the bus for its read tool (StatusTools). Said func(Status) Logf func(string, ...any) Now func() time.Time // Bound is how long it asks before it raises; zero is views.Bound. Wait is the first wait between two asks that // failed, doubling to a minute; zero is five seconds. Bound time.Duration Wait time.Duration mu sync.Mutex values map[string]json.RawMessage st Status asking bool // during are the events heard while an ask runs: taken again after its answer, so none is lost to a picture read // before it was written. during []Event ready chan struct{} } // Status is what a reader's read tool says of one first view (design 25 § 12.2): the owner, the tool and the revision // its picture holds, when it asked and when it was answered, how many asks, the hand acts among them, and whether its // condition is raised. type Status struct { Owner string `json:"owner"` Tool string `json:"tool"` State string `json:"state"` Revision uint64 `json:"revision"` Created time.Time `json:"created,omitzero"` Keys int `json:"keys"` Asked time.Time `json:"asked,omitzero"` Answered time.Time `json:"answered,omitzero"` Asks int `json:"asks"` ByHand int `json:"by-hand"` Raised bool `json:"raised"` // Why is the last reason it asked (start, a gap and which, a hand act) or why the last ask failed. Why string `json:"why,omitempty"` Error string `json:"error,omitempty"` } func (f *Follower) now() time.Time { if f.Now != nil { return f.Now().UTC() } return time.Now().UTC() } func (f *Follower) logf(format string, args ...any) { if f.Logf != nil { f.Logf(format, args...) } } func (f *Follower) init() { if f.values == nil { f.values = map[string]json.RawMessage{} f.ready = make(chan struct{}) f.st = Status{Owner: f.Owner, Tool: f.Tool, State: f.State} } } func (f *Follower) say(s Status) { if f.Said != nil { f.mu.Lock() s.Keys = len(f.values) f.mu.Unlock() f.Said(s) } } // Status is the follower's status now. func (f *Follower) Status() Status { f.mu.Lock() defer f.mu.Unlock() f.init() s := f.st s.Keys = len(f.values) return s } // Ready is closed once a first view was answered. func (f *Follower) Ready() <-chan struct{} { f.mu.Lock() defer f.mu.Unlock() f.init() return f.ready } // Values is a copy of the picture. func (f *Follower) Values() map[string]json.RawMessage { f.mu.Lock() defer f.mu.Unlock() f.init() return maps.Clone(f.values) } // Get is one key of the picture. func (f *Follower) Get(key string) (json.RawMessage, bool) { f.mu.Lock() defer f.mu.Unlock() f.init() v, ok := f.values[key] return v, ok } // Start asks the first view until it is answered or ctx ends (ADR 0615 §6): each failed ask waits, doubling, and past // Bound raises the condition. It returns once answered. func (f *Follower) Start(ctx context.Context) error { return f.askUntil(ctx, "start", false) } // Again asks the first view once more by hand (a re-read tool, recorded by the caller as a hand act), until answered or // ctx ends. func (f *Follower) Again(ctx context.Context, why string) error { return f.askUntil(ctx, "by hand: "+why, true) } func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error { f.mu.Lock() f.init() if f.asking { f.mu.Unlock() return nil // one ask at a time; the one running brings the picture to its newest } f.asking = true f.during = nil f.st.Why = why if byHand { f.st.ByHand++ } f.mu.Unlock() defer func() { f.mu.Lock(); f.asking = false; f.mu.Unlock() }() bound, wait := f.Bound, f.Wait if bound == 0 { bound = Bound } if wait == 0 { wait = 5 * time.Second } first := f.now() for { err := f.askOnce(ctx) if err == nil { return nil } f.mu.Lock() f.st.Error = err.Error() raise := !f.st.Raised && f.now().Sub(first) >= bound if raise { f.st.Raised = true } s := f.st f.mu.Unlock() f.logf("the first view %s of %s's %s is not answered (%v): asked again in %s", f.Tool, f.Owner, f.State, err, wait) if raise && f.Raise != nil { f.Raise(s) } f.say(s) pause, cancel := context.WithTimeout(ctx, wait) // a context's bound between two asks, no timer of its own <-pause.Done() cancel() if ctx.Err() != nil { return ctx.Err() } if wait *= 2; wait > time.Minute { wait = time.Minute } } } // askOnce asks every page and, once whole, takes it as the picture. func (f *Follower) askOnce(ctx context.Context) error { f.mu.Lock() f.st.Asked, f.st.Asks = f.now(), f.st.Asks+1 asked := f.st f.mu.Unlock() f.say(asked) var first Answer values := map[string]json.RawMessage{} after := "" for page := 0; ; page++ { a, err := f.Ask(ctx, after) if err != nil { return err } if page == 0 { first = a } maps.Copy(values, a.Values) if a.Next == "" || a.Next == after { break } after = a.Next } f.mu.Lock() old := f.values f.values = values wasRaised := f.st.Raised f.st.Revision, f.st.Created, f.st.Answered, f.st.Raised, f.st.Error = first.Revision, first.Created, f.now(), false, "" // The events heard while it asked, after the answer's revision and in order, are taken again. during := f.during f.during = nil slices.SortFunc(during, func(a, b Event) int { return cmp.Compare(a.Revision, b.Revision) }) for _, ev := range during { if ev.Created.Equal(f.st.Created) && ev.Revision == f.st.Revision+1 && !ev.Gap { if ev.Deleted { delete(f.values, ev.Key) } else { f.values[ev.Key] = ev.Value } f.st.Revision = ev.Revision } } values = maps.Clone(f.values) s := f.st select { case <-f.ready: default: close(f.ready) } f.mu.Unlock() if wasRaised && f.Clear != nil { f.Clear(s) } f.say(s) if f.Changed != nil { for k, v := range values { if ov, ok := old[k]; !ok || string(ov) != string(v) { f.Changed(k, v, false, time.Time{}) } } for k := range old { if _, ok := values[k]; !ok { f.Changed(k, nil, true, time.Time{}) } } } return nil } // Apply takes one event of the owner's views family into the picture and says whether it was a gap: another Created // (the owner's state made again), a revision past the one after the picture's, or a value too large to carry. A // revision at or below the picture's is already in it (a durable consumer handing what it held) and is passed over. // Before the first view is answered an event is passed over too: the answer holds it. func (f *Follower) Apply(ev Event) (gap bool, why string) { f.mu.Lock() f.init() if f.asking { f.during = append(f.during, ev) } select { case <-f.ready: default: f.mu.Unlock() return false, "" } switch { case !ev.Created.Equal(f.st.Created): why = fmt.Sprintf("%s's %s was made again (or not marked when the first view was answered): created %s, held %s", f.Owner, f.State, ev.Created.Format(time.RFC3339), f.st.Created.Format(time.RFC3339)) case ev.Revision <= f.st.Revision: f.mu.Unlock() return false, "" case ev.Revision > f.st.Revision+1: why = fmt.Sprintf("%s's %s at revision %d, the picture holds %d: %d changes missed", f.Owner, f.State, ev.Revision, f.st.Revision, ev.Revision-f.st.Revision-1) case ev.Gap: why = fmt.Sprintf("%s's %s.%s at revision %d was too large to carry", f.Owner, f.State, ev.Key, ev.Revision) } if why != "" { f.mu.Unlock() return true, why } if ev.Deleted { delete(f.values, ev.Key) } else { f.values[ev.Key] = ev.Value } f.st.Revision = ev.Revision f.mu.Unlock() if f.Changed != nil { f.Changed(ev.Key, ev.Value, ev.Deleted, ev.At) } return false, "" } // Hear is a reader's consumer of one owner's views family, routing each event to the follower of its state; a gap asks // that follower's first view once more on its own goroutine, said in the log. A message it cannot read is terminated. func Hear(ctx context.Context, followers map[string]*Follower, logf func(string, ...any)) func(jetstream.Msg) { return func(m jetstream.Msg) { if err := HearEvent(ctx, followers, m.Data(), logf); err != nil { logf("views: %s is not a views event, dropped: %v", m.Subject(), err) _ = m.Term() return } _ = m.Ack() } } // HearEvent is Hear for one event given as its data, for a reader that takes the owner's events by another path than // a consumer of its own (the tool runner's): routed to the follower of its state, an event of a state it does not // follow passed over, and a gap asked once more on its own goroutine as a gap, never as a hand ask (Gap). Data that is // not a views event is answered as an error, and nothing is asked. func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte, logf func(string, ...any)) error { ev, err := DecodeEvent(data) if err != nil { return err } f := followers[ev.State] if f == nil { return nil // a state this reader does not follow } if gap, why := f.Apply(ev); gap { logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool) go func() { _ = f.Gap(ctx, why) }() } return nil } // Gap asks the first view once more for a gap the reader saw, until answered or ctx ends: counted among its asks, // never among its asks by hand. func (f *Follower) Gap(ctx context.Context, why string) error { return f.askUntil(ctx, "a gap: "+why, false) } // AskVia is a Follower's Ask through a process's call of the owner's tool (process.Conn.Call), decoding its answer. func AskVia(call func(key string, body any) (json.RawMessage, error), tool string) func(context.Context, string) (Answer, error) { return func(_ context.Context, after string) (Answer, error) { body := map[string]any{} if after != "" { body["after"] = after } raw, err := call(tool, body) if err != nil { return Answer{}, err } return DecodeAnswer(raw) } }