149 lines
6.1 KiB
Go
149 lines
6.1 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"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
|
|
}
|
|
stream := declaration.FileStreamPrefix + bucket
|
|
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 the declaration names", 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)
|
|
}
|
|
|
|
var (
|
|
_ FilesBus = OverNATS{}
|
|
_ FilesBus = (*natsLink)(nil)
|
|
)
|