package broker import ( "context" "fmt" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go/jetstream" ) // The compositions' buckets (novox/hq ADR 0539 clauses 2 and 4): one per node, `composed-`, holding that node's // compositions, which only that node's node-engine reads, since a composition can carry a sealed value (ADR 0488 §2's // reason); and `mesh-controller_composed`, holding each composition's digest and the judgement of its move, which the // assigner reads. The controller writes both, by compare-and-set while it holds the lease; each keeps five revisions // of a key. The names and layouts are mesh-sdk's (delivery.ComposedBucketFor, ControllerComposedBucket, ComposedKey). // ControllerComposed is the local name of the compositions' digests, as a reader names it in `reads` after // `mesh-controller.`: delivery.ControllerComposedBucket. const ControllerComposed = "composed" // ComposedValueBytes is the most a composition may hold: a file over 64 KiB travels by digest, so a module's // composition is far smaller; just under the server's largest message, which also carries the subject and headers. const ComposedValueBytes = 1<<20 - 8<<10 // ComposedBucketBytes is a node's composition bucket's size: five revisions of every module's compositions of the node. const ComposedBucketBytes = 256 << 20 // ComposedDigestsBytes is the size of the bucket of digests: a value of a few hundred bytes per node, module and // fingerprint kept. const ComposedDigestsBytes = 64 << 20 // ComposedConfig is a node's composition bucket as the controller asserts it. func ComposedConfig(node string) (jetstream.KeyValueConfig, error) { bucket, err := delivery.ComposedBucketFor(node) if err != nil { return jetstream.KeyValueConfig{}, err } return jetstream.KeyValueConfig{ Bucket: bucket, Description: "the compositions of " + node + " (novox/hq ADR 0539): each module's resources for this node at one " + "fingerprint, and the node's own, signed; written by the controller alone, read by this node's node-engine " + "alone; deleted when outside the module's five newest fingerprints and named by no want", History: delivery.ComposedHistory, MaxValueSize: ComposedValueBytes, MaxBytes: ComposedBucketBytes, Storage: jetstream.FileStorage, }, nil } // ComposedDigestsConfig is the bucket of the compositions' digests as the controller asserts it. func ComposedDigestsConfig() jetstream.KeyValueConfig { return jetstream.KeyValueConfig{ Bucket: delivery.ControllerComposedBucket, Description: "each composition's digest and the judgement of its move from what the node runs (novox/hq ADR 0539 " + "clauses 2 and 14): written by the controller alone, read by the assigner", History: delivery.ControllerComposedHistory, MaxValueSize: 4 << 10, MaxBytes: ComposedDigestsBytes, Storage: jetstream.FileStorage, } } // EnsureComposed creates the bucket of digests and each named node's composition bucket if absent and brings their // options to match. An update, never a delete. func EnsureComposed(ctx context.Context, js jetstream.JetStream, nodes ...string) error { if _, err := js.CreateOrUpdateKeyValue(ctx, ComposedDigestsConfig()); err != nil { return fmt.Errorf("asserting bucket %s: %w", delivery.ControllerComposedBucket, err) } for _, node := range nodes { cfg, err := ComposedConfig(node) if err != nil { return err } if _, err := js.CreateOrUpdateKeyValue(ctx, cfg); err != nil { return fmt.Errorf("asserting bucket %s: %w", cfg.Bucket, err) } } return nil } // ComposedWrites is what the controller publishes to write a node's compositions: the bucket's own subjects. func ComposedWrites(node string) (string, error) { bucket, err := delivery.ComposedBucketFor(node) if err != nil { return "", err } return "$KV." + bucket + ".>", nil } // ComposedReads is what a node's node-engine publishes to follow its wants and compositions (ADR 0539 clauses 1, 2 and // 9): every key of its own composition bucket, and its own wants (`want..` and `want..node`) in the // assigner's bucket; nothing of another node's, nor the bucket of digests. func ComposedReads(node string) ([]string, error) { bucket, err := delivery.ComposedBucketFor(node) if err != nil { return nil, err } if !safeSubject.MatchString(node) { return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node) } return append(readGrants(bucket, ""), readGrants(delivery.AssignerWantsBucket, "want."+node+".>")...), nil } // VerbsTheNodeAgentAsks are the seat verbs a node's node-engine calls of its own machine while it applies a want (ADR // 0539 clause 16, ADR 0429 §14): its backup holder backs a data module up before it moves, and says when it was. Asked // of its own machine only, `mesh.seat..tool..`. var VerbsTheNodeAgentAsks = []SeatVerb{{Seat: "node-backup", Verb: "now"}, {Seat: "node-backup", Verb: "backed-up"}} // NodeAgentAsks are the subjects of VerbsTheNodeAgentAsks for one node. func NodeAgentAsks(node string) ([]string, error) { if !safeSubject.MatchString(node) { return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node) } var out []string for _, v := range VerbsTheNodeAgentAsks { out = append(out, "mesh.seat."+v.Seat+".tool."+v.Verb+"."+node) } return out, nil } // NodeAgentHears are the events a node's node-engine subscribes while it applies a want: the backup holder's word that a // backup is taken (`backed-up`), from whichever module holds node-backup on its machine (restic today). var NodeAgentHears = []string{"mesh.mod." + BackupHolderModule + ".event.backed-up"} // BackupHolderModule is the module that holds node-backup on every node, restic: the one whose word a node-engine hears. // A second holder of the seat is a change of this grant. const BackupHolderModule = "restic" // administers says a person may invoke every tool (`*`): the operator. func administers(invokes []string) bool { for _, i := range invokes { if i == "*" { return true } } return false } // HandPathWrites is what the operator's account publishes for release 6's hand path (ADR 0539 clauses 2 and 10): a want // of any node, written and read back, in the assigner's bucket; each named node's compositions; the digests; and each // node's object bucket for a file over 64 KiB, whose put reads an object's metadata and purges a replaced one's chunks. func HandPathWrites(nodes []string) ([]string, error) { out := append([]string{"$KV." + delivery.AssignerWantsBucket + ".want.>"}, readGrants(delivery.AssignerWantsBucket, "want.>")...) out = append(out, "$KV."+delivery.ControllerComposedBucket+".>") out = append(out, readGrants(delivery.ControllerComposedBucket, "")...) for _, node := range nodes { w, err := ComposedWrites(node) if err != nil { return nil, err } bucket, _ := delivery.ComposedBucketFor(node) out = append(out, w) out = append(out, readGrants(bucket, "")...) files, err := DeclarationFilesWrites(node) if err != nil { return nil, err } stream := "OBJ_" + DeclarationFilesBucket + node out = append(out, files, "$JS.API.STREAM.INFO."+stream, "$JS.API.DIRECT.GET."+stream+".>", "$JS.API.STREAM.MSG.GET."+stream, "$JS.API.STREAM.PURGE."+stream) } return out, nil } // DeclarationsStream is the stream of each node's last declaration, `mesh.node..declare` (issue 107). Release 6 // sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the // switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have // names the new node-engine, and from then on neither asserts it nor any node's consumer on it. const DeclarationsStream = "NODES" // RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the // declarations' stream, and no node's consumer of a declaration. func RaiseAfterDeclarations(r Raiser) error { for _, s := range MeshStreams() { if s.Name == DeclarationsStream { continue } if err := r.EnsureStream(s); err != nil { return fmt.Errorf("asserting stream %s: %w", s.Name, err) } } return AssertMeshConsumers(r) }