jochen
31804bd8c2
A delivery and the five-minute reconcile were two paths that applied, ordered only by a lock, and each order it allowed was met live (issues 257, 261, 267). Now both only enqueue: one worker takes the newest declaration held when it starts, applies it once and makes one report, and reports leave in the order they are made. A declaration may carry the controller's lease epoch beside its sequence; one older than what this node applied is refused before anything is touched, counted, logged and reported. A report carries the declaration's epoch and sequence and the host's own report sequence, kept on disk so it goes on increasing across restarts and self-updates. Without an epoch, today's behaviour stands.
80 lines
2.3 KiB
Go
80 lines
2.3 KiB
Go
package link
|
|
|
|
import (
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type said struct {
|
|
body []byte
|
|
|
|
mu sync.Mutex
|
|
handled bool
|
|
}
|
|
|
|
func (s *said) Body() []byte { return s.body }
|
|
func (s *said) Handled() error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.handled = true
|
|
return nil
|
|
}
|
|
func (s *said) wasHandled() bool {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.handled
|
|
}
|
|
|
|
func arriving(bodies ...string) chan Declaration {
|
|
ch := make(chan Declaration, 8)
|
|
for _, b := range bodies {
|
|
ch <- &said{body: []byte(b)}
|
|
}
|
|
return ch
|
|
}
|
|
|
|
// newest is what the queue's worker would take from what the link gathered behind `first`.
|
|
func newest(arriving <-chan Declaration, first Declaration, window time.Duration) (Declaration, []Declaration) {
|
|
return pick(gather(arriving, first, window))
|
|
}
|
|
|
|
// A machine asked to be five things becomes the last one: what is already waiting supersedes what
|
|
// arrived first, and everything set aside is named so it can be reported.
|
|
func TestWhatIsAlreadyWaitingSupersedesWhatArrivedFirst(t *testing.T) {
|
|
waiting := arriving("two", "three", "four")
|
|
apply, superseded := newest(waiting, &said{body: []byte("one")}, 50*time.Millisecond)
|
|
if string(apply.Body()) != "four" {
|
|
t.Fatalf("applied %q, not the newest", apply.Body())
|
|
}
|
|
if len(superseded) != 3 || string(superseded[0].Body()) != "one" ||
|
|
string(superseded[2].Body()) != "three" {
|
|
t.Fatalf("set aside %d: %v", len(superseded), superseded)
|
|
}
|
|
}
|
|
|
|
// One declaration with nothing behind it is applied as it always was, after the window.
|
|
func TestALoneDeclarationIsAppliedAfterTheWindow(t *testing.T) {
|
|
began := time.Now()
|
|
apply, superseded := newest(arriving(), &said{body: []byte("only")}, 30*time.Millisecond)
|
|
if string(apply.Body()) != "only" || len(superseded) != 0 {
|
|
t.Fatalf("got %q with %d set aside", apply.Body(), len(superseded))
|
|
}
|
|
if time.Since(began) < 30*time.Millisecond {
|
|
t.Fatal("did not wait the window for a straggler")
|
|
}
|
|
}
|
|
|
|
// A straggler within the window is taken; one after it is the next push.
|
|
func TestAStragglerWithinTheWindowIsTaken(t *testing.T) {
|
|
waiting := arriving()
|
|
go func() {
|
|
time.Sleep(20 * time.Millisecond)
|
|
waiting <- &said{body: []byte("late")}
|
|
}()
|
|
apply, superseded := newest(waiting, &said{body: []byte("first")}, 100*time.Millisecond)
|
|
if string(apply.Body()) != "late" || len(superseded) != 1 {
|
|
t.Fatalf("got %q with %d set aside", apply.Body(), len(superseded))
|
|
}
|
|
}
|