diff --git a/go/views/follow.go b/go/views/follow.go index 11107a4..1d2519f 100644 --- a/go/views/follow.go +++ b/go/views/follow.go @@ -51,7 +51,14 @@ type Follower struct { // 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{} + // heardRev is the highest revision of the owner's state (made at heardCreated) any event handed to Apply carried, + // whether it was taken into the picture, passed over or heard while an ask ran: what the picture must reach, so a + // reader that heard it knows it is behind without a further event. + heardRev uint64 + heardCreated time.Time + // catching is set while catchUp runs: one catch-up at a time. + catching bool + 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 @@ -163,7 +170,14 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error f.st.ByHand++ } f.mu.Unlock() - defer func() { f.mu.Lock(); f.asking = false; f.mu.Unlock() }() + released := false + defer func() { + if !released { + f.mu.Lock() + f.asking = false + f.mu.Unlock() + } + }() bound, wait := f.Bound, f.Wait if bound == 0 { bound = Bound @@ -175,6 +189,23 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error for { err := f.askOnce(ctx) if err == nil { + // An event heard that the answer does not hold yet (the owner answered before it wrote it, or the event came + // before the first view and was passed over, or it was a gap heard while this ask ran) leaves the picture + // behind. No further event is waited for: it is asked again on a goroutine of its own, decided in the same + // check that ends this ask, so an event heard after the check finds no ask running and asks itself (ADR 0615 + // §5). The ask that returns has brought the picture as far as the owner answered; the module need not wait. + f.mu.Lock() + behind := f.behindLocked() + f.asking, released = false, true + start := behind != "" && !f.catching + if start { + f.catching = true + } + f.mu.Unlock() + if start { + f.logf("views: %s's %s is behind an event heard (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool) + go f.catchUp(ctx) + } return nil } f.mu.Lock() @@ -190,10 +221,7 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error 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 { + if !f.pause(ctx, wait) { return ctx.Err() } if wait *= 2; wait > time.Minute { @@ -202,6 +230,66 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error } } +// catchUp asks the first view once more, each time after a pause (the reader's Wait, doubling to a minute: an owner that +// answers from before its last write is not asked in a hot loop), until the picture holds every revision heard. It +// holds no ask while it waits, so a gap, an ask by hand or an event heard meanwhile is not held up by it. +func (f *Follower) catchUp(ctx context.Context) { + wait := f.Wait + if wait == 0 { + wait = 5 * time.Second + } + for { + if !f.pause(ctx, wait) { + f.mu.Lock() + f.catching = false + f.mu.Unlock() + return + } + f.mu.Lock() + behind := f.behindLocked() + if behind == "" { + f.catching = false + f.mu.Unlock() + return + } + f.mu.Unlock() + _ = f.askUntil(ctx, "behind an event heard: "+behind, false) + if wait *= 2; wait > time.Minute { + wait = time.Minute + } + } +} + +// pause waits a context's bound between two asks, no timer of its own; false once ctx ended. +func (f *Follower) pause(ctx context.Context, wait time.Duration) bool { + p, cancel := context.WithTimeout(ctx, wait) + <-p.Done() + cancel() + return ctx.Err() == nil +} + +// behindLocked says why the picture is behind an event heard, or "" when it holds every revision heard. The caller +// holds mu. +func (f *Follower) behindLocked() string { + switch { + case f.heardCreated.IsZero(): + return "" + case f.heardCreated.After(f.st.Created): + return fmt.Sprintf("the state was made again at %s, the picture holds the one made at %s", f.heardCreated.Format(time.RFC3339), f.st.Created.Format(time.RFC3339)) + case f.heardCreated.Equal(f.st.Created) && f.heardRev > f.st.Revision: + return fmt.Sprintf("revision %d heard, the picture holds %d", f.heardRev, f.st.Revision) + } + return "" +} + +// Behind says whether an event heard is not in the picture yet, and why: for a reader's tool and a test. +func (f *Follower) Behind() string { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + return f.behindLocked() +} + // askOnce asks every page and, once whole, takes it as the picture. func (f *Follower) askOnce(ctx context.Context) error { f.mu.Lock() @@ -282,6 +370,12 @@ func (f *Follower) askOnce(ctx context.Context) error { func (f *Follower) Apply(ev Event) (gap bool, why string) { f.mu.Lock() f.init() + switch { + case ev.Created.After(f.heardCreated): + f.heardCreated, f.heardRev = ev.Created, ev.Revision + case ev.Created.Equal(f.heardCreated) && ev.Revision > f.heardRev: + f.heardRev = ev.Revision + } if f.asking { f.during = append(f.during, ev) } @@ -312,7 +406,9 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) { f.values[ev.Key] = ev.Value } f.st.Revision = ev.Revision + s := f.st f.mu.Unlock() + f.say(s) // the revision the picture holds follows the events, not only the asks (a read tool says where it stands) if f.Changed != nil { f.Changed(ev.Key, ev.Value, ev.Deleted, ev.At) } diff --git a/go/views/views_test.go b/go/views/views_test.go index 272afb7..4c79227 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strings" + "sync" "testing" "time" @@ -224,6 +225,66 @@ func TestViews(t *testing.T) { t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision) } }) + t.Run("a reader that heard an event its answer does not hold asks again at once, with no further event", func(t *testing.T) { + // The owner's state: revision 9 at first, 12 once its writer is done. The picture a first view reads lags the events. + for name, drop := range map[string]func(f *Follower, owner *uint64){ + // A gap heard while the ask runs: the running ask swallowed the gap's own ask and the event was not consecutive. + "a gap heard while an ask runs": func(f *Follower, owner *uint64) { *owner = 12; f.Apply(ev(12, "e", `{"n":5}`)) }, + // An event heard before the first view, passed over, the owner answering from before its write. + "an event heard before the first view": func(f *Follower, owner *uint64) { f.Apply(ev(10, "e", `{"n":5}`)) }, + } { + t.Run(name, func(t *testing.T) { + owner := uint64(9) + var f *Follower + asks := 0 + f, _ = follower(func(context.Context, string) (Answer, error) { + asks++ + a := Answer{V: Version, Owner: "o", State: "modules", Revision: owner, Created: made, Values: map[string]json.RawMessage{"e": json.RawMessage(`{"n":5}`)}} + if asks == 1 && name == "a gap heard while an ask runs" { + drop(f, &owner) // heard while this ask runs; the answer is the one read before it + } + if asks == 2 { + owner = 12 // the owner has written it by the second ask + a.Revision = owner + } + return a, nil + }) + if name == "an event heard before the first view" { + drop(f, &owner) + } + caught := make(chan struct{}) + var once sync.Once + f.Said = func(s Status) { + if s.Revision >= 10 { + once.Do(func() { close(caught) }) + } + } + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + // No further event follows: the follower asks once more by itself and holds what was heard. + select { + case <-caught: + case <-time.After(10 * time.Second): + t.Fatal("the reader stayed behind the event it heard") + } + if s := f.Status(); s.Revision < 10 || f.Behind() != "" || s.Asks != 2 || !strings.HasPrefix(s.Why, "behind an event heard") { + t.Errorf("revision %d after %d asks (%q), behind: %q", s.Revision, s.Asks, s.Why, f.Behind()) + } + }) + } + }) + t.Run("an event taken into the picture is said in the status, so a read tool shows the revision held", func(t *testing.T) { + var told []uint64 + f, _ := follower(answer) + f.Said = func(s Status) { told = append(told, s.Revision) } + _ = f.Start(context.Background()) + told = nil + f.Apply(ev(10, "c", `{"n":3}`)) + if len(told) != 1 || told[0] != 10 { + t.Errorf("said revisions %v, want [10]", told) + } + }) t.Run("an event heard by its data asks a gap once more as a gap, never by hand", func(t *testing.T) { asked := make(chan struct{}, 4) f, _ := follower(func(ctx context.Context, after string) (Answer, error) {