199 lines
6.9 KiB
Go
199 lines
6.9 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
)
|
|
|
|
// The merge window's clock is on the bus (novox/hq ADR 0407 decision 1, issue 973): an open batch has a key
|
|
// `<batch>.quiet` that lives the window and is made again by each merge into the batch, and a key `<batch>.most`
|
|
// made once when the batch opens that lives the window's most. The first of the two to expire closes the batch,
|
|
// and the watcher hears it: no timer polls the batches. The store still keeps the times (ClosesAt, AtMost) as the
|
|
// record of when a window closed, which `plans` reads; the keys are what wakes the controller.
|
|
|
|
// windowClock is the bus's side of the merge window.
|
|
type windowClock interface {
|
|
// Arm makes a batch's keys: its quiet key again, to live quiet; its most key if it has none, to live most.
|
|
Arm(ctx context.Context, batch string, quiet, most time.Duration) error
|
|
// Forget takes a batch's keys away: it was cut, or closed, and nothing is to wake the controller for it.
|
|
Forget(ctx context.Context, batch string)
|
|
}
|
|
|
|
// windowLook is the look a window's expiry takes: the watcher's, and the fallback timer's.
|
|
var windowLook atomic.Pointer[func(batch string)]
|
|
|
|
var (
|
|
windowMu sync.RWMutex
|
|
windowKeyed windowClock // nil in a process that is not the serving controller
|
|
)
|
|
|
|
// setWindowClock makes clock the merge window's; nil takes it away.
|
|
func setWindowClock(c windowClock) {
|
|
windowMu.Lock()
|
|
defer windowMu.Unlock()
|
|
windowKeyed = c
|
|
}
|
|
|
|
func windowClockOf() windowClock {
|
|
windowMu.RLock()
|
|
defer windowMu.RUnlock()
|
|
return windowKeyed
|
|
}
|
|
|
|
// windowDeadlines are how long a batch's two keys are to live from now: what is left of its window and of its
|
|
// most, each at least a second (the least a key can live). Not armed for a batch whose window is closed already
|
|
// (by a stop, by its time, or because it is cut at once) or that is no batch.
|
|
func windowDeadlines(b inventory.Plan, now time.Time) (quiet, most time.Duration, armed bool) {
|
|
if b.Delivery == nil || b.Delivery.Batch == nil || windowClosed(b, now) {
|
|
return 0, 0, false
|
|
}
|
|
// Rounded up to a whole second, at least one: the server rounds a key's time to live down to whole seconds, so
|
|
// a fraction would close the key before the record's time and the look would cut nothing.
|
|
least := func(d time.Duration) time.Duration {
|
|
if d < time.Second {
|
|
return time.Second
|
|
}
|
|
return (d + time.Second - 1) / time.Second * time.Second
|
|
}
|
|
w := b.Delivery.Batch
|
|
return least(w.ClosesAt.Sub(now)), least(w.AtMost.Sub(now)), true
|
|
}
|
|
|
|
// armWindow puts a batch's clock on the bus after its record was kept. Where the bus will not take it the window
|
|
// closes by a timer of this batch alone, once, as long as is left: never silently never.
|
|
func armWindow(ctx context.Context, b inventory.Plan, now time.Time) {
|
|
c := windowClockOf()
|
|
quiet, most, armed := windowDeadlines(b, now)
|
|
if c == nil || !armed {
|
|
return
|
|
}
|
|
if err := c.Arm(ctx, b.ID, quiet, most); err != nil {
|
|
fmt.Printf("batches: the window of %s could not be put on the bus (%v); it closes by a timer of its own\n", b.ID, err)
|
|
if look := windowLook.Load(); look != nil {
|
|
time.AfterFunc(quiet, func() { (*look)(b.ID) })
|
|
}
|
|
}
|
|
}
|
|
|
|
// busWindow is the clock on the controller's bucket.
|
|
type busWindow struct {
|
|
kv jetstream.KeyValue
|
|
}
|
|
|
|
// newBusWindow opens the bucket and starts the watcher, which names each batch whose key expired to look until
|
|
// the context ends. Only expiry wakes it: a key taken away by Forget or made again by Arm leaves a delete.
|
|
func newBusWindow(ctx context.Context, api jetstream.JetStream, look func(batch string)) (*busWindow, error) {
|
|
kv, err := api.KeyValue(ctx, broker.MergeWindowBucket)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the merge window's bucket: %w", err)
|
|
}
|
|
w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("watching the merge window: %w", err)
|
|
}
|
|
go func() {
|
|
defer w.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
return
|
|
}
|
|
if e == nil || e.Operation() != jetstream.KeyValuePurge {
|
|
continue // a put, or a delete this controller made
|
|
}
|
|
if batch, _, found := strings.Cut(e.Key(), "."); found && look != nil {
|
|
look(batch)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
return &busWindow{kv: kv}, nil
|
|
}
|
|
|
|
func (w *busWindow) Arm(ctx context.Context, batch string, quiet, most time.Duration) error {
|
|
// The most is made once, when the batch opens: it is there already for every merge after the first.
|
|
if _, err := w.kv.Create(ctx, batch+".most", []byte("most"), jetstream.KeyTTL(most)); err != nil &&
|
|
!errors.Is(err, jetstream.ErrKeyExists) {
|
|
return err
|
|
}
|
|
// The quiet key is made again, not changed: a key's time to live is set when it is made.
|
|
if err := w.kv.Delete(ctx, batch+".quiet"); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return err
|
|
}
|
|
if _, err := w.kv.Create(ctx, batch+".quiet", []byte("quiet"), jetstream.KeyTTL(quiet)); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (w *busWindow) Forget(ctx context.Context, batch string) {
|
|
for _, key := range []string{batch + ".quiet", batch + ".most"} {
|
|
_ = w.kv.Delete(ctx, key)
|
|
}
|
|
}
|
|
|
|
// lookAtBatches is the look an expired window asks for: the plans held, as a merge heard holds them, and the batches
|
|
// whose window closed cut (cutBatchesHeld). Waiting for the plans rather than giving up: an expiry is heard once.
|
|
func lookAtBatches(ctx context.Context, open *stores) {
|
|
release, err := open.inventory.HoldPlans(ctx, true)
|
|
if err != nil {
|
|
fmt.Printf("batches: a window closed and the plans could not be held for the look: %v\n", err)
|
|
return
|
|
}
|
|
defer release()
|
|
if err := cutBatchesHeld(ctx, open, time.Now().UTC()); err != nil {
|
|
fmt.Printf("batches: %v\n", err)
|
|
}
|
|
}
|
|
|
|
// startMergeWindow puts the merge window's clock on the bus for as long as the context lives: the controller is
|
|
// the lease holder here, the only one that writes it. The watcher looks at the batches when a key expires; the
|
|
// batches open now get their keys, and one look is taken, so a window that closed while no controller served is
|
|
// cut (ADR 0407 decision 1). No timer polls them.
|
|
func startMergeWindow(ctx context.Context, open *stores, js *broker.JetStream) error {
|
|
api, err := jetstream.New(js.Conn())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
look := func(string) { lookAtBatches(ctx, open) }
|
|
windowLook.Store(&look)
|
|
clock, err := newBusWindow(ctx, api, look)
|
|
if err != nil {
|
|
windowLook.Store(nil)
|
|
return err
|
|
}
|
|
setWindowClock(clock)
|
|
go func() {
|
|
<-ctx.Done()
|
|
setWindowClock(nil)
|
|
windowLook.Store(nil)
|
|
}()
|
|
release, err := open.inventory.HoldPlans(ctx, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer release()
|
|
batches, err := open.inventory.Batches(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
now := time.Now().UTC()
|
|
for _, b := range batches {
|
|
armWindow(ctx, b, now)
|
|
}
|
|
return cutBatchesHeld(ctx, open, now)
|
|
}
|