287 lines
10 KiB
Go
287 lines
10 KiB
Go
package views
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// Emitter is an owner's process saying each change of its own states on its views family (ADR 0615 §5): it follows
|
|
// each state it is given from the revision its mark says it said up to, publishes one Event per change, and writes the
|
|
// mark once it has said every change the bus had handed it. A part watches only its own state (ADR 0596 §4): an owner
|
|
// follows its own buckets, never another's.
|
|
type Emitter struct {
|
|
// Owner is the owner module, said in every event.
|
|
Owner string
|
|
// Subject is where an event of the given local name lands on the owner's own stream (process.Account.StreamSubject).
|
|
Subject func(event string) string
|
|
// Publish sends one message to the owner's stream, its id the bus's dedup id.
|
|
Publish func(ctx context.Context, msg *nats.Msg, id string) error
|
|
// Marks is the owner's state MarksState; nil says nothing is marked, and every start says from the bucket's end.
|
|
Marks jetstream.KeyValue
|
|
// TooLarge is told of a value too large for an event (said as a gap): the owner's finding naming the key. Nil
|
|
// only logs it.
|
|
TooLarge func(state, key string, size int)
|
|
// Findings writes and clears the owner's finding of a change not said past UnsaidAfter (UnsaidKind, its own state
|
|
// `findings`); nil says it in the log only.
|
|
Findings FindingWriter
|
|
// UnsaidAfter is how long a change may stay not said before it is the owner's finding; zero is Bound.
|
|
UnsaidAfter time.Duration
|
|
// Wait pauses before a change not said is said again (its attempt so far); it answers false when ctx ended. Nil is
|
|
// a backoff from 250 ms doubling to 30 s.
|
|
Wait func(ctx context.Context, attempt int) bool
|
|
Logf func(string, ...any)
|
|
Now func() time.Time
|
|
}
|
|
|
|
// JetStreamPublish is Emitter.Publish through js, with the bus's dedup id.
|
|
func JetStreamPublish(js jetstream.JetStream) func(context.Context, *nats.Msg, string) error {
|
|
return func(ctx context.Context, msg *nats.Msg, id string) error {
|
|
pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
|
defer cancel()
|
|
_, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id))
|
|
return err
|
|
}
|
|
}
|
|
|
|
func (e Emitter) now() time.Time {
|
|
if e.Now != nil {
|
|
return e.Now().UTC()
|
|
}
|
|
return time.Now().UTC()
|
|
}
|
|
|
|
func (e Emitter) logf(format string, args ...any) {
|
|
if e.Logf != nil {
|
|
e.Logf(format, args...)
|
|
}
|
|
}
|
|
|
|
// created is when the bucket's stream was made, and its fill now.
|
|
func created(ctx context.Context, kv jetstream.KeyValue) (time.Time, uint64, Fill, error) {
|
|
st, err := kv.Status(ctx)
|
|
if err != nil {
|
|
return time.Time{}, 0, Fill{}, err
|
|
}
|
|
b, ok := st.(*jetstream.KeyValueBucketStatus)
|
|
if !ok || b.StreamInfo() == nil {
|
|
return time.Time{}, 0, Fill{}, fmt.Errorf("the bus answered no stream behind %s", st.Bucket())
|
|
}
|
|
info := b.StreamInfo()
|
|
return info.Created.UTC(), info.State.LastSeq, Fill{Keys: uint64(info.State.NumSubjects), Bytes: info.State.Bytes, MaxBytes: info.Config.MaxBytes}, nil
|
|
}
|
|
|
|
// Start follows state's bucket kv: from the revision after its mark when the mark is of the same bucket (the same
|
|
// Created), otherwise from the bucket's end, writing a mark there first, so a reader's first view taken now misses
|
|
// nothing. It returns once the watch is set; the changes are said on a goroutine until ctx ends, and a watch that ends
|
|
// is sent on errs.
|
|
func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, errs chan<- error) error {
|
|
made, last, fill, err := created(ctx, kv)
|
|
if err != nil {
|
|
return fmt.Errorf("the state %s cannot be followed for its views: %w", state, err)
|
|
}
|
|
from := last + 1
|
|
if m, ok := e.mark(ctx, state); ok && m.Created.Equal(made) && m.Revision <= last {
|
|
from = m.Revision + 1
|
|
} else {
|
|
e.writeMark(ctx, state, Mark{State: state, Revision: last, Created: made, Fill: fill})
|
|
}
|
|
w, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(from))
|
|
if err != nil {
|
|
return fmt.Errorf("the state %s cannot be watched for its views: %w", state, err)
|
|
}
|
|
fillNow := func() (Fill, error) {
|
|
_, _, fill, err := created(ctx, kv)
|
|
return fill, err
|
|
}
|
|
go func() {
|
|
defer w.Stop()
|
|
e.follow(ctx, state, made, w.Updates(), fillNow, errs)
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
// follow says each change handed on updates in order, and marks once every change handed so far is said. A change
|
|
// that fails to be said is said again, the same change on a backoff (Wait), until it is said or ctx ends: the mark
|
|
// never moves past a change not said, so no reader misses one (design 25 §12.2).
|
|
func (e Emitter) follow(ctx context.Context, state string, made time.Time, updates <-chan jetstream.KeyValueEntry, fillNow func() (Fill, error), errs chan<- error) {
|
|
said := uint64(0)
|
|
raised := false
|
|
// A finding a run before this one raised is not known to this one: the first change said clears it once, so a
|
|
// restart never leaves it standing (the finding is about the state, its key the same).
|
|
leftover := e.Findings != nil
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case u, ok := <-updates:
|
|
if !ok {
|
|
if errs != nil {
|
|
errs <- fmt.Errorf("the watch of %s for its views ended", state)
|
|
}
|
|
return
|
|
}
|
|
if u != nil {
|
|
var first time.Time
|
|
for attempt := 1; ; attempt++ {
|
|
err := e.say(ctx, state, made, u)
|
|
if err == nil {
|
|
break
|
|
}
|
|
e.logf("views: %s.%s at %d is not said (try %d): %v; said again before the next change", state, u.Key(), u.Revision(), attempt, err)
|
|
if first.IsZero() {
|
|
first = e.now()
|
|
}
|
|
if !raised && e.now().Sub(first) >= e.unsaidAfter() {
|
|
raised = e.raiseUnsaid(state, u, first, err)
|
|
}
|
|
if !e.wait(ctx, attempt) {
|
|
// Not marked: the next start says it again from the mark, and the bus keeps it once by its id.
|
|
return
|
|
}
|
|
}
|
|
if raised || leftover {
|
|
raised = !e.clearUnsaid(state)
|
|
leftover = false
|
|
}
|
|
said = u.Revision()
|
|
}
|
|
// Marked once every change handed so far is said, so a busy state writes one mark per burst.
|
|
if said > 0 && len(updates) == 0 {
|
|
fill, err := fillNow()
|
|
if err != nil {
|
|
e.logf("views: the fill of %s is not read: %v", state, err)
|
|
}
|
|
e.writeMark(ctx, state, Mark{State: state, Revision: said, Created: made, Fill: fill})
|
|
said = 0
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e Emitter) unsaidAfter() time.Duration {
|
|
if e.UnsaidAfter > 0 {
|
|
return e.UnsaidAfter
|
|
}
|
|
return Bound
|
|
}
|
|
|
|
// raiseUnsaid writes the owner's finding of u not said since first; it answers whether the finding stands.
|
|
func (e Emitter) raiseUnsaid(state string, u jetstream.KeyValueEntry, first time.Time, cause error) bool {
|
|
e.logf("views: %s.%s at %d is not said since %s: the owner's finding %s", state, u.Key(), u.Revision(), first.Format(time.RFC3339), UnsaidKind)
|
|
if e.Findings == nil {
|
|
return false
|
|
}
|
|
if err := e.Findings.Put(UnsaidFinding(e.Owner, state, u.Key(), u.Revision(), first, cause.Error(), e.now())); err != nil {
|
|
e.logf("views: the finding of %s's change not said could not be written: %v", state, err)
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// clearUnsaid clears the finding once the change is said; it answers whether it is cleared.
|
|
func (e Emitter) clearUnsaid(state string) bool {
|
|
if err := e.Findings.Delete(UnsaidFinding(e.Owner, state, "", 0, time.Time{}, "", e.now())); err != nil {
|
|
e.logf("views: the finding of %s's change not said could not be cleared: %v", state, err)
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// wait is the pause before the attempt+1'th try of a change not said: Wait when given, else a backoff from 250 ms
|
|
// doubling to 30 s. It answers false once ctx ends.
|
|
func (e Emitter) wait(ctx context.Context, attempt int) bool {
|
|
if e.Wait != nil {
|
|
return e.Wait(ctx, attempt)
|
|
}
|
|
d := 250 * time.Millisecond
|
|
for i := 1; i < attempt && d < 30*time.Second; i++ {
|
|
d *= 2
|
|
}
|
|
d = min(d, 30*time.Second)
|
|
t := time.NewTimer(d)
|
|
defer t.Stop()
|
|
select {
|
|
case <-ctx.Done():
|
|
return false
|
|
case <-t.C:
|
|
return true
|
|
}
|
|
}
|
|
|
|
// Event is the event saying u, a change of state made at made.
|
|
func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) Event {
|
|
ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made, At: u.Created().UTC()}
|
|
if u.Operation() != jetstream.KeyValuePut {
|
|
ev.Deleted = true
|
|
return ev
|
|
}
|
|
if n := len(u.Value()); n > MaxValueBytes {
|
|
ev.Gap = true
|
|
e.logf("views: %s.%s at %d holds %d bytes, more than an event carries: said as a gap, so a reader asks its first view", state, u.Key(), u.Revision(), n)
|
|
if e.TooLarge != nil {
|
|
e.TooLarge(state, u.Key(), n)
|
|
}
|
|
return ev
|
|
}
|
|
ev.Value = json.RawMessage(u.Value())
|
|
if !json.Valid(ev.Value) {
|
|
ev.Value, _ = json.Marshal(string(u.Value()))
|
|
}
|
|
return ev
|
|
}
|
|
|
|
func (e Emitter) say(ctx context.Context, state string, made time.Time, u jetstream.KeyValueEntry) error {
|
|
ev := e.Event(state, made, u)
|
|
data, err := ev.Encode()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
msg := nats.NewMsg(e.Subject(EventName(state)))
|
|
msg.Header.Set("x-event-id", ev.ID())
|
|
msg.Header.Set("x-source", e.Owner)
|
|
msg.Header.Set("x-time", e.now().Format(time.RFC3339Nano))
|
|
msg.Header.Set("content-type", "application/json")
|
|
msg.Data = data
|
|
return e.Publish(ctx, msg, ev.ID())
|
|
}
|
|
|
|
func (e Emitter) mark(ctx context.Context, state string) (Mark, bool) {
|
|
if e.Marks == nil {
|
|
return Mark{}, false
|
|
}
|
|
got, err := e.Marks.Get(ctx, state)
|
|
if err != nil {
|
|
if !errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err)
|
|
}
|
|
return Mark{}, false
|
|
}
|
|
m, err := DecodeMark(got.Value())
|
|
if err != nil {
|
|
e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err)
|
|
return Mark{}, false
|
|
}
|
|
return m, true
|
|
}
|
|
|
|
func (e Emitter) writeMark(ctx context.Context, state string, m Mark) {
|
|
if e.Marks == nil {
|
|
return
|
|
}
|
|
m.V, m.At = Version, e.now()
|
|
data, err := json.Marshal(m)
|
|
if err != nil {
|
|
e.logf("views: the mark of %s is not written: %v", state, err)
|
|
return
|
|
}
|
|
if _, err := e.Marks.Put(ctx, state, data); err != nil {
|
|
e.logf("views: the mark of %s at %d is not written (%v): a first view answers the one before", state, m.Revision, err)
|
|
}
|
|
}
|