220 lines
6.9 KiB
Go
220 lines
6.9 KiB
Go
package keeper
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"nats-tools/internal/bussnap"
|
|
)
|
|
|
|
// JetStream is the bus as the keeper's own account reaches it (Bus): each object a stream, a key-value bucket or an
|
|
// object store of JetStream, read, created, grown and deleted by its stream.
|
|
type JetStream struct {
|
|
JS jetstream.JetStream
|
|
// Snap takes a snapshot of streams and says where it is kept (bussnap through the keeper's own connection).
|
|
Snap func(ctx context.Context, streams []string) (string, error)
|
|
// NC and Dir are the connection and the scratch directory the keeper's restore in place of a snapshot uses (design 47 G §681).
|
|
NC *nats.Conn
|
|
Dir string
|
|
Log func(string, ...any)
|
|
}
|
|
|
|
// Held reads the object's stream.
|
|
func (b JetStream) Held(ctx context.Context, o Object) (Settings, Fill, bool, error) {
|
|
s, err := b.JS.Stream(ctx, o.Stream())
|
|
if errors.Is(err, jetstream.ErrStreamNotFound) {
|
|
return Settings{}, Fill{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return Settings{}, Fill{}, false, retryable(err)
|
|
}
|
|
info, err := s.Info(ctx)
|
|
if err != nil {
|
|
return Settings{}, Fill{}, false, retryable(err)
|
|
}
|
|
return SettingsOf(o.Kind, info.Config), Fill{Bytes: info.State.Bytes, Messages: info.State.Msgs}, true, nil
|
|
}
|
|
|
|
// SettingsOf reads a stream's configuration as the settings an object of kind is asked with.
|
|
func SettingsOf(kind string, c jetstream.StreamConfig) Settings {
|
|
s := Settings{MaxBytes: c.MaxBytes, MaxAgeSeconds: int64(c.MaxAge / time.Second), Discard: DiscardOld,
|
|
WorkQueue: c.Retention == jetstream.WorkQueuePolicy, DuplicatesSeconds: int64(c.Duplicates / time.Second)}
|
|
if c.Discard == jetstream.DiscardNew {
|
|
s.Discard = DiscardNew
|
|
}
|
|
if c.MaxMsgSize > 0 {
|
|
s.MaxMsgSize = int64(c.MaxMsgSize)
|
|
}
|
|
switch kind {
|
|
case KindBucket:
|
|
s.History = c.MaxMsgsPerSubject
|
|
s.KeyTTL = c.AllowMsgTTL
|
|
s.DuplicatesSeconds = 0
|
|
case KindObjects:
|
|
s.DuplicatesSeconds = 0
|
|
default:
|
|
if c.MaxMsgsPerSubject > 0 {
|
|
s.History = c.MaxMsgsPerSubject
|
|
}
|
|
if !s.WorkQueue {
|
|
s.DuplicatesSeconds = 0
|
|
}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// Create creates the object; a refusal of the account's grant is ErrRetry: the grants part composes the keeper's grant
|
|
// to create a module's names from the same accepted manifests, and may not have yet.
|
|
func (b JetStream) Create(ctx context.Context, o Object) error {
|
|
var err error
|
|
switch o.Kind {
|
|
case KindBucket:
|
|
_, err = b.JS.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: o.Name, History: uint8(o.Settings.History),
|
|
TTL: time.Duration(o.Settings.MaxAgeSeconds) * time.Second, MaxBytes: o.Settings.MaxBytes,
|
|
MaxValueSize: int32(o.Settings.MaxMsgSize), Storage: jetstream.FileStorage, Replicas: 1,
|
|
LimitMarkerTTL: limitMarker(o.Settings.KeyTTL)})
|
|
case KindObjects:
|
|
_, err = b.JS.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: o.Name, MaxBytes: o.Settings.MaxBytes,
|
|
TTL: time.Duration(o.Settings.MaxAgeSeconds) * time.Second, Storage: jetstream.FileStorage, Replicas: 1})
|
|
default:
|
|
_, err = b.JS.CreateStream(ctx, streamConfig(o))
|
|
}
|
|
return retryable(err)
|
|
}
|
|
|
|
func limitMarker(on bool) time.Duration {
|
|
if on {
|
|
return time.Second
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func streamConfig(o Object) jetstream.StreamConfig {
|
|
c := jetstream.StreamConfig{Name: o.Name, Subjects: o.Subjects, Retention: jetstream.LimitsPolicy,
|
|
MaxBytes: o.Settings.MaxBytes, MaxAge: time.Duration(o.Settings.MaxAgeSeconds) * time.Second,
|
|
Discard: jetstream.DiscardOld, Storage: jetstream.FileStorage, Replicas: 1, MaxMsgs: -1, MaxMsgsPerSubject: -1}
|
|
if o.Settings.Discard == DiscardNew {
|
|
c.Discard = jetstream.DiscardNew
|
|
}
|
|
if o.Settings.WorkQueue {
|
|
c.Retention = jetstream.WorkQueuePolicy
|
|
c.Duplicates = time.Duration(o.Settings.DuplicatesSeconds) * time.Second
|
|
}
|
|
if o.Settings.MaxMsgSize > 0 {
|
|
c.MaxMsgSize = int32(o.Settings.MaxMsgSize)
|
|
}
|
|
if o.Settings.History > 0 {
|
|
c.MaxMsgsPerSubject = o.Settings.History
|
|
}
|
|
return c
|
|
}
|
|
|
|
// Update brings the held stream to the object's settings, keeping every other part of its configuration.
|
|
func (b JetStream) Update(ctx context.Context, o Object) error {
|
|
s, err := b.JS.Stream(ctx, o.Stream())
|
|
if err != nil {
|
|
return retryable(err)
|
|
}
|
|
info, err := s.Info(ctx)
|
|
if err != nil {
|
|
return retryable(err)
|
|
}
|
|
c := info.Config
|
|
c.MaxBytes = o.Settings.MaxBytes
|
|
c.MaxAge = time.Duration(o.Settings.MaxAgeSeconds) * time.Second
|
|
if o.Settings.History > 0 {
|
|
c.MaxMsgsPerSubject = o.Settings.History
|
|
}
|
|
_, err = b.JS.UpdateStream(ctx, c)
|
|
return retryable(err)
|
|
}
|
|
|
|
// Delete deletes the object's stream.
|
|
func (b JetStream) Delete(ctx context.Context, o Object) error {
|
|
err := b.JS.DeleteStream(ctx, o.Stream())
|
|
if errors.Is(err, jetstream.ErrStreamNotFound) {
|
|
return nil
|
|
}
|
|
return retryable(err)
|
|
}
|
|
|
|
// Snapshot takes the streams' snapshot.
|
|
func (b JetStream) Snapshot(ctx context.Context, streams []string) (string, error) {
|
|
if b.Snap == nil {
|
|
return "", fmt.Errorf("no snapshot is taken here, so nothing is removed")
|
|
}
|
|
return b.Snap(ctx, streams)
|
|
}
|
|
|
|
// retryable marks a refusal of the account's permissions, or a bus that answers nothing, as ErrRetry.
|
|
func retryable(err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
words := strings.ToLower(err.Error())
|
|
if strings.Contains(words, "permission") || strings.Contains(words, "authorization") || errors.Is(err, context.DeadlineExceeded) ||
|
|
strings.Contains(words, "no responders") || strings.Contains(words, "timeout") {
|
|
return fmt.Errorf("%w: %v", ErrRetry, err)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// RestrictedMetadata is the key a restricted stream's metadata says it under (ADR 0363 §4): the grants part writes it.
|
|
const RestrictedMetadata = "mesh-restricted"
|
|
|
|
// Restricted reads the streams' metadata: true when any says it is restricted.
|
|
func (b JetStream) Restricted(ctx context.Context, streams []string) (bool, error) {
|
|
for _, name := range streams {
|
|
s, err := b.JS.Stream(ctx, name)
|
|
if err != nil {
|
|
return false, retryable(err)
|
|
}
|
|
info, err := s.Info(ctx)
|
|
if err != nil {
|
|
return false, retryable(err)
|
|
}
|
|
if _, ok := info.Config.Metadata[RestrictedMetadata]; ok {
|
|
return true, nil
|
|
}
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
// SnapshotStreams reads the streams a snapshot file holds from its manifest.
|
|
func (b JetStream) SnapshotStreams(path string) ([]string, error) {
|
|
m, err := bussnap.ManifestOf(path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out []string
|
|
for _, s := range m.Streams {
|
|
out = append(out, s.Name)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// Live says whether a stream of that name exists.
|
|
func (b JetStream) Live(ctx context.Context, name string) (bool, error) {
|
|
return bussnap.Live(ctx, b.NC, name)
|
|
}
|
|
|
|
// Restore puts the snapshot's streams back under their own names.
|
|
func (b JetStream) Restore(ctx context.Context, path string) error {
|
|
log := b.Log
|
|
if log == nil {
|
|
log = func(string, ...any) {}
|
|
}
|
|
return bussnap.RestoreFile(ctx, b.NC, path, b.Dir, log)
|
|
}
|
|
|
|
// DeleteStream removes the stream by its name.
|
|
func (b JetStream) DeleteStream(ctx context.Context, name string) error {
|
|
return bussnap.DeleteStream(ctx, b.NC, name)
|
|
}
|