jschoubben
882e75c954
On the node whose compositions carry the bus's list, the engine also watches mesh-grants' key user-list.sealed. While it says follow grants, the object it names is read by name, opened with this machine's sealing key, checked against the key's digest and the guard, and written to the same file at 0600 only when its digest differs; the bus reloads it itself. The have says which list is followed and this machine's sealing key, which the grants part seals to. The follower's test waits for its watch to catch up before a later put, which raced under load. Pins mesh-sdk 175a5567.
197 lines
7.9 KiB
Go
197 lines
7.9 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// FilesBus is a link that can read an object of the bucket the controller keeps a declaration's large files in
|
|
// (novox/hq ADR 0486 §1, §2). Separate from Bus, as HaveBus is: a link that cannot is still a link, and a declaration
|
|
// whose files are given by digest is then refused whole, naming that.
|
|
type FilesBus interface {
|
|
// FileObject is the object of that name, no larger than limit bytes, read within ctx. **Reads by name only**: a
|
|
// node-engine's account may neither list the bucket nor write it (ADR 0486 §2).
|
|
FileObject(ctx context.Context, name string, limit int64) ([]byte, error)
|
|
}
|
|
|
|
type filesKey struct{}
|
|
|
|
// WithFiles is ctx carrying the link a declaration's file references are fetched on.
|
|
func WithFiles(ctx context.Context, bus any) context.Context {
|
|
if f, ok := bus.(FilesBus); ok {
|
|
return context.WithValue(ctx, filesKey{}, f)
|
|
}
|
|
return ctx
|
|
}
|
|
|
|
// FilesFrom is the fetch the apply's context carries, or nil when the node has no such link.
|
|
func FilesFrom(ctx context.Context) func(context.Context, string, int64) ([]byte, error) {
|
|
if f, ok := ctx.Value(filesKey{}).(FilesBus); ok {
|
|
return f.FileObject
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// objectInfo is the part of an object store's metadata record this read needs (nats.go's ObjectInfo, as the store writes it).
|
|
type objectInfo struct {
|
|
Name string `json:"name"`
|
|
NUID string `json:"nuid"`
|
|
Size uint64 `json:"size"`
|
|
Digest string `json:"digest"`
|
|
Deleted bool `json:"deleted"`
|
|
}
|
|
|
|
// FileObject reads one object by name from this node's own object bucket (declaration-files-<node>), without listing the bucket or looking at
|
|
// anything but the object it names (novox/hq ADR 0486 §2: a node-engine's account may read by name and may neither
|
|
// list nor write).
|
|
//
|
|
// **The bus subjects this uses, and no others** (the grant of issue 1697 is exactly these; reviewer's note 5 there), for
|
|
// bucket B = declaration-files-<this node>, stream S = OBJ_B, and E the base64 (URL alphabet) of the object's name:
|
|
//
|
|
// request $JS.API.DIRECT.GET.S.$O.B.M.E the object's metadata. Grant only as ...DIRECT.GET.S.$O.B.M.*, never
|
|
// the bare ...DIRECT.GET.S, which takes a sequence number
|
|
// request $JS.API.CONSUMER.CREATE.S.<hash>.$O.B.C.<nuid>
|
|
// an ordered consumer on the object's chunks. Grant only as
|
|
// ...CREATE.S.*.$O.B.C.*, never with ">" (a consumer filtered on
|
|
// $O.B.M.> would read every object's name). Repeated on a reset
|
|
// subscribe a new inbox under the connection's own inbox prefix, where the chunks are delivered
|
|
// publish $JS.FC.S.> the flow-control replies of the ordered consumer; without them a
|
|
// large object stalls at the flow-control window
|
|
// request $JS.API.CONSUMER.DELETE.S.<hash> on unsubscribe; best effort, the consumer's inactivity threshold
|
|
// removes it if refused
|
|
//
|
|
// It calls no STREAM.INFO and no STREAM.MSG.GET: js.ObjectStore asks for the stream's info (which lists every object's name)
|
|
// and store.Get reads the metadata with a message get a grant cannot keep from reading by sequence.
|
|
func (b OverNATS) FileObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
|
js := b.JS
|
|
wait := 10 * time.Second
|
|
if at, ok := ctx.Deadline(); ok {
|
|
if wait = time.Until(at); wait <= 0 {
|
|
return nil, context.DeadlineExceeded
|
|
}
|
|
}
|
|
if b.Conn != nil {
|
|
bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
js = bounded
|
|
}
|
|
if b.Node == "" {
|
|
return nil, errors.New("this link does not know its node's name, so it cannot name the node's own bucket")
|
|
}
|
|
bucket, err := declaration.FileBucketFor(b.Node)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return objectByName(ctx, js, bucket, declaration.FileStreamPrefix+bucket, name, limit)
|
|
}
|
|
|
|
// GrantsObject reads one object of the grants part's bucket of sealed user lists by its name (novox/hq ADR 0547 clause 7),
|
|
// with the same subjects and nothing more as FileObject: the bus node is granted exactly those (mesh-sdk
|
|
// buscompose.GrantsUserListFollows).
|
|
func (b OverNATS) GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
|
js := b.JS
|
|
wait := 10 * time.Second
|
|
if at, ok := ctx.Deadline(); ok {
|
|
if wait = time.Until(at); wait <= 0 {
|
|
return nil, context.DeadlineExceeded
|
|
}
|
|
}
|
|
if b.Conn != nil {
|
|
bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
js = bounded
|
|
}
|
|
return objectByName(ctx, js, delivery.GrantsUserListObjects, delivery.GrantsUserListObjectStream, name, limit)
|
|
}
|
|
|
|
// objectByName is one object of bucket (its stream stream) by its name, no larger than limit, verified against its digest.
|
|
func objectByName(ctx context.Context, js nats.JetStreamContext, bucket, stream, name string, limit int64) ([]byte, error) {
|
|
meta, err := js.GetLastMsg(stream, "$O."+bucket+".M."+base64.URLEncoding.EncodeToString([]byte(name)),
|
|
nats.DirectGet(), nats.Context(ctx))
|
|
if err != nil {
|
|
if errors.Is(err, nats.ErrMsgNotFound) {
|
|
return nil, fmt.Errorf("no object %s in the bucket %s", name, bucket)
|
|
}
|
|
return nil, fmt.Errorf("the metadata of object %s: %w", name, err)
|
|
}
|
|
var info objectInfo
|
|
if err := json.Unmarshal(meta.Data, &info); err != nil {
|
|
return nil, fmt.Errorf("the metadata of object %s is not readable: %w", name, err)
|
|
}
|
|
if info.Deleted {
|
|
return nil, fmt.Errorf("the object %s was deleted from the bucket %s", name, bucket)
|
|
}
|
|
if info.Size > uint64(limit) {
|
|
return nil, fmt.Errorf("the object is larger than the %d bytes allowed", limit)
|
|
}
|
|
if info.Size == 0 {
|
|
return []byte{}, nil
|
|
}
|
|
sub, err := js.SubscribeSync("$O."+bucket+".C."+info.NUID,
|
|
nats.OrderedConsumer(), nats.BindStream(stream))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("reading the chunks of object %s: %w", name, err)
|
|
}
|
|
defer sub.Unsubscribe()
|
|
body := make([]byte, 0, info.Size)
|
|
for uint64(len(body)) < info.Size {
|
|
msg, err := sub.NextMsgWithContext(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("reading the chunks of object %s (%d of %d bytes): %w", name, len(body), info.Size, err)
|
|
}
|
|
if uint64(len(body))+uint64(len(msg.Data)) > info.Size {
|
|
return nil, fmt.Errorf("the chunks of object %s are more than the %d bytes its metadata says", name, info.Size)
|
|
}
|
|
body = append(body, msg.Data...)
|
|
}
|
|
sum := sha256.Sum256(body)
|
|
if want := "SHA-256=" + base64.URLEncoding.EncodeToString(sum[:]); info.Digest != "" && info.Digest != want {
|
|
return nil, fmt.Errorf("the chunks of object %s do not hash to the digest of its metadata", name)
|
|
}
|
|
return body, nil
|
|
}
|
|
|
|
func (l *natsLink) FileObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
|
return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.FileObject(ctx, name, limit)
|
|
}
|
|
|
|
func (l *natsLink) GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
|
return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.GrantsObject(ctx, name, limit)
|
|
}
|
|
|
|
// GrantsObjects is a link that can read an object of the grants part's bucket of sealed user lists.
|
|
type GrantsObjects interface {
|
|
GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error)
|
|
}
|
|
|
|
// GrantsObject reads an object of the grants part's bucket by its name on the link open now.
|
|
func (q *Session) GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
|
q.init()
|
|
q.mu.Lock()
|
|
bus := q.bus
|
|
q.mu.Unlock()
|
|
g, ok := bus.(GrantsObjects)
|
|
if bus == nil || !ok {
|
|
return nil, ErrNoLink
|
|
}
|
|
return g.GrantsObject(ctx, name, limit)
|
|
}
|
|
|
|
var (
|
|
_ FilesBus = OverNATS{}
|
|
_ FilesBus = (*natsLink)(nil)
|
|
)
|