359 lines
10 KiB
Go
359 lines
10 KiB
Go
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 (<module>.<tool>).
|
|
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) {
|
|
ev, err := DecodeEvent(m.Data())
|
|
if err != nil {
|
|
logf("views: %s is not a views event, dropped: %v", m.Subject(), err)
|
|
_ = m.Term()
|
|
return
|
|
}
|
|
f := followers[ev.State]
|
|
if f == nil {
|
|
_ = m.Ack() // a state this reader does not follow
|
|
return
|
|
}
|
|
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.mu.Lock()
|
|
f.init()
|
|
f.st.Why = "a gap: " + why
|
|
f.mu.Unlock()
|
|
_ = f.askUntil(ctx, "a gap: "+why, false)
|
|
}()
|
|
}
|
|
_ = m.Ack()
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
}
|