299 lines
8.2 KiB
Go
299 lines
8.2 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// KVUpdate is one change of a key in a bucket the node-engine watches (novox/hq ADR 0539 §9). Caught marks the end of
|
|
// the keys that were there when the watch began.
|
|
type KVUpdate struct {
|
|
Key string
|
|
Value []byte
|
|
Revision uint64
|
|
Deleted bool
|
|
Caught bool
|
|
}
|
|
|
|
// KVBus is a link that can watch a bucket's keys: the node agent follows its wants and its compositions with it, and
|
|
// never polls.
|
|
type KVBus interface {
|
|
// KVWatch yields the keys of the bucket matching the pattern now, a Caught update, then every change, until ctx ends
|
|
// or the link does (the channel is then closed). A bucket the node's account may not read is an error.
|
|
KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error)
|
|
}
|
|
|
|
// KVWatch is the watch on the link open now.
|
|
func (b OverNATS) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
|
|
js := b.JS
|
|
if b.Conn != nil {
|
|
bound, err := b.Conn.JetStream()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
js = bound
|
|
}
|
|
kv, err := js.KeyValue(bucket)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
|
|
}
|
|
w, err := kv.Watch(pattern, nats.Context(ctx))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("watching %s in %s: %w", pattern, bucket, err)
|
|
}
|
|
out := make(chan KVUpdate, 64)
|
|
go func() {
|
|
defer close(out)
|
|
defer func() { _ = w.Stop() }()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
return
|
|
}
|
|
var u KVUpdate
|
|
if e == nil {
|
|
u.Caught = true
|
|
} else {
|
|
u = KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
|
|
Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}
|
|
}
|
|
select {
|
|
case out <- u:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
return out, nil
|
|
}
|
|
|
|
func (l *natsLink) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
|
|
return OverNATS{Conn: l.conn, JS: l.js}.KVWatch(ctx, bucket, pattern)
|
|
}
|
|
|
|
var (
|
|
_ KVBus = OverNATS{}
|
|
_ KVBus = (*natsLink)(nil)
|
|
)
|
|
|
|
// KVHistoryBus is a link that can read the revisions a bucket keeps of one key: the node agent takes the composition its
|
|
// want names from them when a newer one replaced it (novox/hq ADR 0539 clause 1).
|
|
type KVHistoryBus interface {
|
|
// KVHistory is the key's revisions, oldest first; none when the key has none.
|
|
KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error)
|
|
}
|
|
|
|
// historyWithin bounds one read of a key's revisions.
|
|
const historyWithin = 10 * time.Second
|
|
|
|
// KVHistory reads the key's revisions on the link open now.
|
|
func (b OverNATS) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
|
|
js := b.JS
|
|
if b.Conn != nil {
|
|
bound, err := b.Conn.JetStream()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
js = bound
|
|
}
|
|
kv, err := js.KeyValue(bucket)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
|
|
}
|
|
// Bounded on its own: a read that does not answer must not hold the node agent (the review of issue 1945).
|
|
ctx, cancel := context.WithTimeout(ctx, historyWithin)
|
|
defer cancel()
|
|
entries, err := kv.History(key, nats.Context(ctx))
|
|
if errors.Is(err, nats.ErrKeyNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, err)
|
|
}
|
|
out := make([]KVUpdate, 0, len(entries))
|
|
for _, e := range entries {
|
|
out = append(out, KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
|
|
Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (l *natsLink) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
|
|
return OverNATS{Conn: l.conn, JS: l.js}.KVHistory(ctx, bucket, key)
|
|
}
|
|
|
|
var (
|
|
_ KVHistoryBus = OverNATS{}
|
|
_ KVHistoryBus = (*natsLink)(nil)
|
|
)
|
|
|
|
// KVHistory reads a key's revisions on the link open now.
|
|
func (q *Session) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
|
|
q.init()
|
|
q.mu.Lock()
|
|
bus := q.bus
|
|
q.mu.Unlock()
|
|
h, ok := bus.(KVHistoryBus)
|
|
if bus == nil || !ok {
|
|
return nil, ErrNoLink
|
|
}
|
|
return h.KVHistory(ctx, bucket, key)
|
|
}
|
|
|
|
// ErrNoLink is what a watch asks for with no link open.
|
|
var ErrNoLink = errors.New("no link to the bus is open")
|
|
|
|
// KVWatch watches a bucket on the link open now. It ends when ctx does or the link closes.
|
|
func (q *Session) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
|
|
q.init()
|
|
q.mu.Lock()
|
|
bus := q.bus
|
|
q.mu.Unlock()
|
|
w, ok := bus.(KVBus)
|
|
if bus == nil || !ok {
|
|
return nil, ErrNoLink
|
|
}
|
|
return w.KVWatch(ctx, bucket, pattern)
|
|
}
|
|
|
|
// WhenLinked waits for a link to be open and returns a context that ends when that link does (or ctx does): what the
|
|
// node agent runs under, so its watches end with the link and begin again with the next.
|
|
func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.CancelFunc, error) {
|
|
q.init()
|
|
for {
|
|
q.mu.Lock()
|
|
if q.bus != nil {
|
|
gone := q.gone
|
|
q.mu.Unlock()
|
|
linked, cancel := context.WithCancel(ctx)
|
|
go func() {
|
|
select {
|
|
case <-gone:
|
|
cancel()
|
|
case <-linked.Done():
|
|
}
|
|
}()
|
|
return linked, cancel, nil
|
|
}
|
|
changed := q.changed
|
|
q.mu.Unlock()
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, nil, ctx.Err()
|
|
case <-changed:
|
|
}
|
|
}
|
|
}
|
|
|
|
// EventBus is a link that can hear an event on a subject of core NATS: the node agent waits for a backup's end with it,
|
|
// and never polls.
|
|
type EventBus interface {
|
|
// Hear yields the payloads published on the subject until stop is called or the link ends.
|
|
Hear(subject string) (events <-chan []byte, stop func(), err error)
|
|
}
|
|
|
|
func (b OverNATS) Hear(subject string) (<-chan []byte, func(), error) {
|
|
out := make(chan []byte, 16)
|
|
sub, err := b.Conn.Subscribe(subject, func(m *nats.Msg) {
|
|
select {
|
|
case out <- m.Data:
|
|
default:
|
|
}
|
|
})
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if err := b.Conn.Flush(); err != nil {
|
|
_ = sub.Unsubscribe()
|
|
return nil, nil, err
|
|
}
|
|
return out, func() { _ = sub.Unsubscribe() }, nil
|
|
}
|
|
|
|
func (l *natsLink) Hear(subject string) (<-chan []byte, func(), error) {
|
|
return OverNATS{Conn: l.conn, JS: l.js}.Hear(subject)
|
|
}
|
|
|
|
var (
|
|
_ EventBus = OverNATS{}
|
|
_ EventBus = (*natsLink)(nil)
|
|
)
|
|
|
|
// Hear listens for an event on the link open now, from before the caller asks for what it waits on.
|
|
func (q *Session) Hear(subject string) (<-chan []byte, func(), error) {
|
|
q.init()
|
|
q.mu.Lock()
|
|
bus := q.bus
|
|
q.mu.Unlock()
|
|
h, ok := bus.(EventBus)
|
|
if bus == nil || !ok {
|
|
return nil, nil, ErrNoLink
|
|
}
|
|
return h.Hear(subject)
|
|
}
|
|
|
|
// CreatedAdvisory is the subject the bus says a bucket's stream was created on (the server's own advisory; the same
|
|
// subject mesh-sdk's process.AwaitBucket binds on, and the grants part gives a principal for each bucket it may read).
|
|
func CreatedAdvisory(bucket string) string {
|
|
return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket
|
|
}
|
|
|
|
// KVWatchAwait is KVWatch for a bucket that may not be on the bus yet (the placement seat's states, made when their
|
|
// holder starts): the channel is returned at once, and while the bucket is missing it yields nothing and waiting is told
|
|
// why; when the bus says the bucket's stream was created the watch is made and its updates follow. No timer: the
|
|
// advisory is heard before the first try, so a creation between the try and the wait is not missed. Any other error of
|
|
// the first try is the caller's, as for KVWatch: a bucket the account may not read is not waited for.
|
|
func (q *Session) KVWatchAwait(ctx context.Context, bucket, pattern string, waiting func(error)) (<-chan KVUpdate, error) {
|
|
created, stop, err := q.Hear(CreatedAdvisory(bucket))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
in, err := q.KVWatch(ctx, bucket, pattern)
|
|
if err == nil {
|
|
stop()
|
|
return in, nil
|
|
}
|
|
if !errors.Is(err, nats.ErrBucketNotFound) {
|
|
stop()
|
|
return nil, err
|
|
}
|
|
waiting(err)
|
|
out := make(chan KVUpdate, 64)
|
|
go func() {
|
|
defer close(out)
|
|
defer stop()
|
|
for in == nil {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case _, ok := <-created:
|
|
if !ok {
|
|
return
|
|
}
|
|
w, err := q.KVWatch(ctx, bucket, pattern)
|
|
if err != nil {
|
|
waiting(err)
|
|
continue
|
|
}
|
|
in = w
|
|
}
|
|
}
|
|
for u := range in {
|
|
select {
|
|
case out <- u:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return out, nil
|
|
}
|