A reader that heard an event its picture lacks asks again by itself, and says the revision it holds as events arrive (issue 1966)
A follower passed over an event heard before its first view and dropped a gap heard while an ask ran, so a quiet owner left the reader behind until a further event. It now keeps the highest revision heard and, when an ask ends short of it, asks again on a pause of its own until it holds it. The status a read tool shows followed only the asks, so it said a reader was behind that was not; it follows the events too. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
+102
-6
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user