jschoubben
a092ee9ff9
The bus node can only take over a list that exists: each pass composes the accounts block with the vault's hashes, seals it to the key the bus node says in its have, puts it in the part's own object bucket by digest and writes the key with digest, revision, user count, users left out and which list to follow. take-over is refused, naming why, while no list is sealed to the bus node's current key or the loaded list does not grant the bus node its read. Review of fix/1945-takeover-blockers: no responders on the router's record is an error, not 'no record'; an answer whose record cannot be reached is not closed; an ask that failed to publish is taken back at the router; an error acting on an answer is logged. Pins mesh-sdk 175a5567.
56 lines
1.7 KiB
Go
56 lines
1.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// sealedObjects is the part's object bucket of user lists sealed to the bus node (novox/hq ADR 0547 clause 7), created on
|
|
// its first put: the composer grants the part its create, its puts and deletes, and nothing of another bucket.
|
|
type sealedObjects struct {
|
|
ctx context.Context
|
|
js jetstream.JetStream
|
|
}
|
|
|
|
func (o sealedObjects) store(ctx context.Context) (jetstream.ObjectStore, error) {
|
|
s, err := o.js.ObjectStore(ctx, delivery.GrantsUserListObjects)
|
|
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
|
s, err = o.js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: delivery.GrantsUserListObjects,
|
|
Description: "the bus's user list composed by the grants part, each sealed to the bus node's key under the hex of its digest (novox/hq ADR 0547 clause 7): written by mesh-grants alone, read by name by the node-engine of the node holding mesh-broker",
|
|
Storage: jetstream.FileStorage, MaxBytes: 64 << 20})
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the object bucket %s: %w", delivery.GrantsUserListObjects, err)
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
func (o sealedObjects) PutObject(name string, data []byte) error {
|
|
ctx, cancel := context.WithTimeout(o.ctx, requestWithin)
|
|
defer cancel()
|
|
s, err := o.store(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = s.PutBytes(ctx, name, data)
|
|
return err
|
|
}
|
|
|
|
func (o sealedObjects) DeleteObject(name string) error {
|
|
ctx, cancel := context.WithTimeout(o.ctx, requestWithin)
|
|
defer cancel()
|
|
s, err := o.store(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := s.Delete(ctx, name); err != nil && !errors.Is(err, jetstream.ErrObjectNotFound) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|