jschoubben
3ae2d6e17c
The switch to release 6 is today's controller's last plan, and the bus runs the user list it composes until the new controller writes one: so the new controller's writes of each node's compositions and their digests, each new node-engine's reads of its own compositions and wants and its backup holder's verbs and word, and the assigner's read of the digests must be granted by this controller already (ADR 0539 clauses 2, 6, 9 and 16; the review's A1).
491 lines
27 KiB
Go
491 lines
27 KiB
Go
package broker
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// The controller's own key-value buckets (novox/hq to-be 45 §1, §6, §7).
|
|
//
|
|
// **What the controller must remember across its own restart, it keeps on the bus.** A call's
|
|
// outcome lived in the memory of the process that served it (novox/hq issue 265), so a controller
|
|
// replaced while a push ran answered "no such call" for the one thing its caller had been told to
|
|
// ask about. The bus already outlives the controller and is the shape ADR 0201 gives a module's
|
|
// current state: one value per key, written by one owner, read by anybody granted it. These are the
|
|
// controller's, written by it alone — the writers table of to-be 45 §1 — and asserted on every start
|
|
// like the streams, so a bus raised from nothing has them before the first call is served.
|
|
|
|
// CallsBucket keeps every call of the mesh's own verbs and what came of it; HandActsBucket every act
|
|
// a person did by hand, with why; ConditionsBucket every condition open now (to-be 45 §2), one key
|
|
// each, and ConditionHistoryBucket every transition of one — raised, changed, silenced, cleared —
|
|
// for ninety days.
|
|
//
|
|
// **The history is a bucket of its own** because its keys expire and an open condition's must not:
|
|
// a bucket has one age for every key, and a condition open longer than the history is kept would
|
|
// otherwise vanish from the store while still true.
|
|
var (
|
|
CallsBucket = BucketName(ControllerSeat, "calls")
|
|
HandActsBucket = BucketName(ControllerSeat, "hand-acts")
|
|
ConditionsBucket = BucketName(ControllerSeat, "conditions")
|
|
ConditionHistoryBucket = BucketName(ControllerSeat, "condition-history")
|
|
// LeaseBucket holds the controller's lease (to-be 45 §6): one key, `holder`, which the instance
|
|
// allowed to act writes by compare-and-set and renews; its revision when taken is the epoch.
|
|
LeaseBucket = BucketName(ControllerSeat, "lease")
|
|
// AskedBucket keeps what the controller asked the operator about its conditions (novox/hq ADR 0259):
|
|
// each ask by its id, its options and the actions they stand for, how it ended and whether the
|
|
// controller acted on its warrant — so a restart neither asks twice nor acts twice.
|
|
AskedBucket = BucketName(ControllerSeat, "asked")
|
|
// MergeWindowBucket holds the clock of every open batch's merge window (novox/hq ADR 0407 decision 1): per
|
|
// batch a key `<batch>.quiet`, made again by each merge into it and living the window, and a key
|
|
// `<batch>.most`, made once when it opens and living the window's most. The first to expire closes the
|
|
// batch, and the controller's watcher hears the delete marker. Written by the lease holder alone.
|
|
MergeWindowBucket = BucketName(ControllerSeat, "merge-window")
|
|
// RolloutSilentBucket holds the wait of every machine a rollout has sent a tier to (novox/hq ADR 0406 §5, issue
|
|
// 974): per send a key `<plan>.<tier>.<node>` that lives ApplySilentAfter, made when the controller sends that
|
|
// machine and deleted by the machine's report on that send. Only a missing report lets it expire, and the
|
|
// controller's watcher hearing the expiry ends the wait and names the machine silent. Written by the lease
|
|
// holder alone.
|
|
RolloutSilentBucket = BucketName(ControllerSeat, "rollout-silent")
|
|
// DeliveryMergesBucket holds the merges mesh-delivery announced it made, each with the ask whose Approve it merged
|
|
// on (novox/hq ADR 0420 §7, issue 1025): per merge a key `<owner>/<repo>/<number>/<head sha>` holding the
|
|
// announcement, kept DeliveryMergesKeptFor. What lets the build of a moved module register it from its new
|
|
// repository: the controller reads the ask's warrant from the router and accepts only the operator's Approve of
|
|
// those heads. Written by the controller alone, from the announcement, when it holds the lease.
|
|
DeliveryMergesBucket = BucketName(ControllerSeat, "delivery-merges")
|
|
)
|
|
|
|
// DeliveryMergesKeptFor is how long a merge mesh-delivery announced is kept: a week, as the events are.
|
|
const DeliveryMergesKeptFor = 7 * 24 * time.Hour
|
|
|
|
// The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of
|
|
// record, one key per record, written from the store by the lease holder only when the record changed, with
|
|
// one event on the controller's seat per write. Read by whatever names them in `reads` as
|
|
// `mesh-controller.<mirror>` (ADR 0201 §1), written by nobody else. **No secret in either** (ADR 0201): an
|
|
// assignment's value names its module and node; a module's declared config is what its manifest, which is
|
|
// public, declares.
|
|
//
|
|
// AssignmentsBucket holds which module is assigned to which node, keyed `<node>.<module>`; and
|
|
// DeclaredConfigBucket each registered module's config declarations as its manifest holds them, keyed by
|
|
// the module: what mesh-config resolves a config value from (ADR 0348 clauses 2 and 4); and PredefinedBucket the
|
|
// list of predefined keys the mesh published (the newest trunk build's of the catalogue, ADR 0348 §7), one key
|
|
// `list`: what mesh-config publishes as mesh-config_predefined (design 57 K3, issue 662).
|
|
var (
|
|
AssignmentsBucket = BucketName(ControllerSeat, MirrorAssignments)
|
|
DeclaredConfigBucket = BucketName(ControllerSeat, MirrorDeclaredConfig)
|
|
PredefinedBucket = BucketName(ControllerSeat, MirrorPredefined)
|
|
// DeclaredPackagesBucket holds, keyed by the node, each package the modules assigned to it declare, present or absent,
|
|
// with the declaring modules: what the pacman module's verbs read to say a package is declared (ADR 0507 §5).
|
|
DeclaredPackagesBucket = BucketName(ControllerSeat, MirrorDeclaredPackages)
|
|
)
|
|
|
|
// The mirrors' local names, as a reader names them in `reads` after `mesh-controller.`.
|
|
const (
|
|
MirrorAssignments = "assignments"
|
|
MirrorDeclaredConfig = "declared-config"
|
|
MirrorPredefined = "predefined"
|
|
// MirrorDeclaredPackages is the local name of the packages each node's modules declare (ADR 0507 §5, issue 1840/S3).
|
|
MirrorDeclaredPackages = "declared-packages"
|
|
)
|
|
|
|
// The chain's buckets (novox/hq ADR 0408, issue 1089): builds, rollouts and assignments are one chain kept on the bus
|
|
// alone, one bucket each, one key per record, written by the lease holder by compare-and-set. The assignments bucket is
|
|
// AssignmentsBucket above, which keeps its name and its readers (ADR 0408 §3). Nothing is removed from them (§4), they keep
|
|
// ChainHistory earlier revisions of every key (§5), and a value is never cut (§6).
|
|
var (
|
|
BuildsBucket = BucketName(ControllerSeat, ChainBuilds)
|
|
RolloutsBucket = BucketName(ControllerSeat, ChainRollouts)
|
|
)
|
|
|
|
// **The assignments bucket keeps the marker age (MirrorMarkerTTL) it was made with as a mirror.** A stream that allows
|
|
// messages an age cannot be made to disallow them (the server refuses the update: "message TTL status can not be
|
|
// disabled", nats-server stream.go), so asserting the bucket without the age would fail the controller's start on a
|
|
// mesh that has the bucket. The chain never purges or removes a key (ADR 0408 §4), so no marker is ever written and the
|
|
// age ends nothing; it is kept only so the stream's configuration stays one the server accepts.
|
|
//
|
|
// The chain buckets' local names, as a reader names them in `reads` after `mesh-controller.`.
|
|
const (
|
|
ChainBuilds = "builds"
|
|
ChainRollouts = "rollouts"
|
|
)
|
|
|
|
// The chain buckets' bounds. ChainHistory is the bus's largest history (ADR 0408 §5). A value is at most ChainValueBytes,
|
|
// just under the server's largest message of a mebibyte, which also carries the subject and the headers (ADR 0408 §6):
|
|
// the controller writes none larger and the move refuses to. The buckets' size is the arithmetic of the survey of
|
|
// issue 981: 8585 builds of 1 to 40 KB, 1049 rollouts of about 2 KB a revision and 64 revisions, so 512 MiB for the two
|
|
// that grow, and the assignments, 285 of under a kilobyte, 64 MiB.
|
|
const (
|
|
ChainHistory = 64
|
|
ChainValueBytes = 1<<20 - 8<<10
|
|
ChainBuildsBytes = 512 << 20
|
|
ChainRolloutsBytes = 512 << 20
|
|
ChainAssignmentBytes = 64 << 20
|
|
)
|
|
|
|
// GrantsBucket holds, keyed `<module>.<fingerprint hex>`, each fingerprint of a module whose bus credential the controller
|
|
// has issued (novox/hq ADR 0528 clause 2, ADR 0429 §6 rule 6, issue 1394): written by the controller alone once the bus's
|
|
// user list composed with the module's manifest at that fingerprint was sent to the bus's machine; read by the assigner,
|
|
// whose sixth rule moves a node's want to a fingerprint only once its grant exists. The name and layout are mesh-sdk's
|
|
// (delivery.ControllerGrantsBucket, ControllerGrantKey).
|
|
var GrantsBucket = BucketName(ControllerSeat, ControllerGrants)
|
|
|
|
// ControllerGrants is the grants bucket's local name, as a reader names it in `reads` after `mesh-controller.`.
|
|
const ControllerGrants = "grants"
|
|
|
|
// ChainBuckets are the three buckets of the chain: builds, rollouts and assignments.
|
|
func ChainBuckets() []string { return []string{BuildsBucket, RolloutsBucket, AssignmentsBucket} }
|
|
|
|
// MirrorsWrittenByAnother are the controller's mirrors whose writer is not the controller, each with its reason: the rule
|
|
// that a mirror is written by the controller alone (TestTheMirrorsListAndTheWritersTableAgree) has these exceptions and no
|
|
// other.
|
|
var MirrorsWrittenByAnother = map[string]string{}
|
|
|
|
// The mirrors' caps (design 54 §5): one value per key, no age, 64 KiB a value, 64 MiB a bucket.
|
|
const (
|
|
MirrorValueBytes = 64 << 10
|
|
MirrorBytes = 64 << 20
|
|
// MirrorMarkerTTL is the longest a purge's marker is kept (design 54 §2): the bucket allows a
|
|
// per-message age for it, and the writer gives each marker a day.
|
|
MirrorMarkerTTL = 24 * time.Hour
|
|
)
|
|
|
|
// MergeWindowMarkerTTL is how long the marker an expired window key leaves is kept: long enough for a watcher
|
|
// that is up to hear it, short enough to leave nothing behind.
|
|
const MergeWindowMarkerTTL = time.Minute
|
|
|
|
// RolloutSilentMarkerTTL is how long the marker an expired silent key leaves is kept, for the same reason.
|
|
const RolloutSilentMarkerTTL = time.Minute
|
|
|
|
// RolloutSilentKeptFor is the age of the silent keys' bucket: every report deletes a key and the delete leaves a marker, which
|
|
// with one value per key would stay for ever and fill the bucket. A key lives a quarter of an hour at most, so an hour
|
|
// takes nothing live.
|
|
const RolloutSilentKeptFor = time.Hour
|
|
|
|
// AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own.
|
|
const AskedKeptFor = 30 * 24 * time.Hour
|
|
|
|
// LeaseTTL is how long the lease's key lives unrenewed (to-be 45 §6): fifteen seconds, renewed
|
|
// every five. The bucket's age, so the bus forgets a holder that stopped renewing.
|
|
const LeaseTTL = 15 * time.Second
|
|
|
|
// The bounds to-be 45 §6 sets for calls: the last thousand, or fourteen days, whichever is fewer.
|
|
// A call is two keys — its record, and its answer apart so a listing does not read every answer —
|
|
// so the stream holds twice as many messages as it keeps calls.
|
|
const (
|
|
KeptCallsDurably = 1000
|
|
CallsKeptFor = 14 * 24 * time.Hour
|
|
// CallAnswerBytes is the most of one answer kept: a whole declaration is far smaller, and an
|
|
// answer larger is cut and says so.
|
|
CallAnswerBytes = 64 << 10
|
|
// HandActsKeptFor is as long as a condition's history (to-be 45 §2): an act by hand is read
|
|
// back beside what it addressed.
|
|
HandActsKeptFor = 90 * 24 * time.Hour
|
|
// ConditionHistoryKeptFor is how long a condition's transitions are kept (to-be 45 §2).
|
|
ConditionHistoryKeptFor = 90 * 24 * time.Hour
|
|
)
|
|
|
|
// IsControllerBucket says a bucket is the controller's own, not a module's state nothing declares.
|
|
func IsControllerBucket(bucket string) bool {
|
|
return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket ||
|
|
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket ||
|
|
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket ||
|
|
bucket == DeclaredPackagesBucket ||
|
|
bucket == MergeWindowBucket || bucket == RolloutSilentBucket ||
|
|
bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket ||
|
|
bucket == delivery.HaveBucket || bucket == StatesAnnouncedBucket ||
|
|
bucket == GrantsBucket || bucket == ControllerComposedBucket
|
|
}
|
|
|
|
// ControllerBuckets are the controller's own buckets, in the order they are asserted.
|
|
func ControllerBuckets() []string {
|
|
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket,
|
|
AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, DeclaredPackagesBucket, MergeWindowBucket, RolloutSilentBucket,
|
|
DeliveryMergesBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket, StatesAnnouncedBucket, GrantsBucket,
|
|
ControllerComposedBucket}
|
|
}
|
|
|
|
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
|
|
type ControllerBucketsAsserter interface {
|
|
EnsureControllerBuckets() error
|
|
}
|
|
|
|
// EnsureControllerBuckets creates the controller's buckets if absent and brings their options to
|
|
// match. An update, never a delete: what they hold is the record of what the mesh was asked.
|
|
func (j *JetStream) EnsureControllerBuckets() error {
|
|
js, err := jetstream.New(j.conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
if err := EnsureLeaseBucket(ctx, js); err != nil {
|
|
return err
|
|
}
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: CallsBucket,
|
|
Description: "the calls of the mesh's own verbs and what came of each (novox/hq to-be 45 §6, issue " +
|
|
"265): written by the controller alone, read through `calls`; the last thousand, or fourteen days",
|
|
History: 1,
|
|
TTL: CallsKeptFor,
|
|
MaxValueSize: CallAnswerBytes + 4<<10,
|
|
MaxBytes: 2 * KeptCallsDurably * (CallAnswerBytes + 4<<10),
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", CallsBucket, err)
|
|
}
|
|
// **The count, on the stream under the bucket.** A bucket has an age and a size and no count;
|
|
// the stream it is made of does, and with one value per key the oldest message is the oldest
|
|
// call. Asserted after the bucket, every time, because asserting the bucket writes the stream's
|
|
// configuration whole and puts the count back to none.
|
|
stream, err := js.Stream(ctx, "KV_"+CallsBucket)
|
|
if err != nil {
|
|
return fmt.Errorf("reading the stream under %s: %w", CallsBucket, err)
|
|
}
|
|
cfg := stream.CachedInfo().Config
|
|
if cfg.MaxMsgs != 2*KeptCallsDurably {
|
|
cfg.MaxMsgs = 2 * KeptCallsDurably
|
|
cfg.Discard = jetstream.DiscardOld
|
|
if _, err := js.UpdateStream(ctx, cfg); err != nil {
|
|
return fmt.Errorf("bounding %s to the last %d calls: %w", CallsBucket, KeptCallsDurably, err)
|
|
}
|
|
}
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: HandActsBucket,
|
|
Description: "every act a person did by hand, with why (novox/hq to-be 45 §7): written by the " +
|
|
"controller's repairing verbs and `hand-act record`, read through `hand-acts`",
|
|
History: 1,
|
|
TTL: HandActsKeptFor,
|
|
MaxValueSize: 16 << 10,
|
|
MaxBytes: 64 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", HandActsBucket, err)
|
|
}
|
|
// **No age on the open conditions.** A condition is removed when observation clears it and at no
|
|
// other moment: one that expired would be a fault the store forgot while it was still true.
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: ConditionsBucket,
|
|
Description: "every condition open now, one key each (novox/hq to-be 45 §2): written by the " +
|
|
"controller alone, raised and cleared by observation, read through `conditions`",
|
|
History: 1,
|
|
MaxValueSize: 64 << 10,
|
|
MaxBytes: 64 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", ConditionsBucket, err)
|
|
}
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: ConditionHistoryBucket,
|
|
Description: "every transition of a condition — raised, changed, silenced, cleared — kept ninety " +
|
|
"days (novox/hq to-be 45 §2): written by the controller alone, read through `conditions history`",
|
|
History: 1,
|
|
TTL: ConditionHistoryKeptFor,
|
|
MaxValueSize: 64 << 10,
|
|
MaxBytes: 256 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", ConditionHistoryBucket, err)
|
|
}
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: AskedBucket,
|
|
Description: "what the controller asked the operator about its conditions, and what came of each (novox/hq " +
|
|
"ADR 0259): written by the controller alone; an ask acted on is acted on once",
|
|
History: 1,
|
|
TTL: AskedKeptFor,
|
|
MaxValueSize: 32 << 10,
|
|
MaxBytes: 32 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err)
|
|
}
|
|
// **The merge window: keys that expire.** A key's own time to live is the window; the marker an expired key
|
|
// leaves is what a watcher hears, kept a minute (ADR 0407 decision 1).
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: MergeWindowBucket,
|
|
Description: "the clock of every open batch's merge window (novox/hq ADR 0407): `<batch>.quiet` lives the " +
|
|
"window and is made again by each merge, `<batch>.most` lives the most; the first to expire closes the " +
|
|
"batch; written by the controller alone",
|
|
History: 1,
|
|
MaxValueSize: 4 << 10,
|
|
MaxBytes: 8 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
LimitMarkerTTL: MergeWindowMarkerTTL,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", MergeWindowBucket, err)
|
|
}
|
|
// **The rollout's silent keys: keys that expire** (ADR 0406 §5, issue 974). A key lives as long as the machine
|
|
// may stay silent; the machine's report deletes it, and a marker of expiry is what the watcher hears.
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: RolloutSilentBucket,
|
|
Description: "the wait of every machine a rollout sent a tier to (novox/hq ADR 0406): `<plan>.<tier>.<node>` " +
|
|
"lives the time a machine may stay silent and is deleted by its report; one that expires names the " +
|
|
"machine silent; written by the controller alone",
|
|
History: 1,
|
|
TTL: RolloutSilentKeptFor,
|
|
MaxValueSize: 4 << 10,
|
|
MaxBytes: 8 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
LimitMarkerTTL: RolloutSilentMarkerTTL,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", RolloutSilentBucket, err)
|
|
}
|
|
// **The compositions' digests** (ADR 0539 clauses 2 and 6), asserted ahead of release 6: what its assigner reads.
|
|
if err := EnsureComposedDigests(ctx, js); err != nil {
|
|
return err
|
|
}
|
|
// **The states each module announced** (novox/hq ADR 0495 clauses 2 and 3, issue 1790): one key a module, its last word.
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: StatesAnnouncedBucket,
|
|
Description: "the states each module announced it keeps and which of them others may read (novox/hq ADR 0495): " +
|
|
"`<module>` holds its last announcement; what a reader's `reads` is granted from; written by the controller alone",
|
|
History: 1,
|
|
MaxValueSize: 64 << 10,
|
|
MaxBytes: 16 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", StatesAnnouncedBucket, err)
|
|
}
|
|
// **The grants issued: no age** (ADR 0528 clause 2, issue 1394). One key a module's fingerprint, never removed: a grant
|
|
// that expired would hold back a want whose credential exists.
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: GrantsBucket,
|
|
Description: "each fingerprint of a module whose bus credential the controller issued (novox/hq ADR 0528): " +
|
|
"`<module>.<fingerprint hex>` holds the accounts issued, put once the bus's user list composed with the module's " +
|
|
"manifest at it was sent; read by the assigner through `reads`; written by the controller alone",
|
|
History: delivery.ControllerGrantsHistory,
|
|
MaxValueSize: 4 << 10,
|
|
MaxBytes: 64 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", GrantsBucket, err)
|
|
}
|
|
// **The merges mesh-delivery announced: keys that age** (ADR 0420 §7, issue 1025). One message a key; a week.
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: DeliveryMergesBucket,
|
|
Description: "the merges mesh-delivery announced it made, each with the ask whose Approve it merged on (novox/hq " +
|
|
"ADR 0420): `<owner>/<repo>/<number>/<head sha>` holds the announcement, kept a week; what lets the build " +
|
|
"of a moved module register it from its new repository; written by the controller alone",
|
|
History: 1,
|
|
TTL: DeliveryMergesKeptFor,
|
|
MaxValueSize: 16 << 10,
|
|
MaxBytes: 16 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", DeliveryMergesBucket, err)
|
|
}
|
|
// **The chain: no age, no removal, history 64, no cut** (ADR 0408 §4 to §6). A key is never removed, so no purge
|
|
// marker needs an age and a key that expired would be a record the bus forgot.
|
|
for _, c := range []struct {
|
|
bucket, why string
|
|
bytes int64
|
|
}{
|
|
{BuildsBucket, "every build, one key each by its id, with its repository, commit, reuse key and what it " +
|
|
"published (novox/hq ADR 0408, design 47): written by the controller alone, never removed, read through `reads`",
|
|
ChainBuildsBytes},
|
|
{RolloutsBucket, "every rollout, one key each by its id: the commit it fixed per repository, its tiers, the " +
|
|
"builds it carried and the assignments it changed (novox/hq ADR 0408, design 47): written by the " +
|
|
"controller alone, never removed, read through `reads`", ChainRolloutsBytes},
|
|
} {
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, chainBucket(c.bucket, c.why, c.bytes, 0)); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", c.bucket, err)
|
|
}
|
|
}
|
|
// **The have bucket: what each machine has of each module** (ADR 0429 §1, issue 1136). Twenty revisions of each key and
|
|
// no age. Asserted by the controller, written by each machine's node-engine under its own node's keys and by nobody
|
|
// else: the controller's grant leaves it out.
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, HaveBucketConfig()); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", delivery.HaveBucket, err)
|
|
}
|
|
// **The assignments keep the marker age they had as a mirror.** A stream that allows messages an age cannot be
|
|
// made to disallow them (the server refuses: "message TTL status can not be disabled"), so the bucket that was
|
|
// the assignments mirror is asserted with the age it was made with. Nothing is purged from it, so no marker is ever
|
|
// written and the age ends nothing.
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, chainBucket(AssignmentsBucket,
|
|
"which module is assigned to which node, keyed <node>.<module>, with the build, commit and rollout it runs "+
|
|
"there (novox/hq ADR 0408, design 47): written by the controller alone, never removed, read through `reads`",
|
|
ChainAssignmentBytes, MirrorMarkerTTL)); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", AssignmentsBucket, err)
|
|
}
|
|
// **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose
|
|
// marker lives a day; a key that expired would be a record the bus forgot while the store still holds it.
|
|
for _, m := range []struct{ bucket, why string }{
|
|
{DeclaredConfigBucket, "each registered module's config declarations as its manifest holds them, keyed by " +
|
|
"the module (novox/hq ADR 0348, design 57 F3): written by the controller alone from its store when a " +
|
|
"registration changes them, read through `reads`"},
|
|
{PredefinedBucket, "the list of predefined keys the mesh published, the newest trunk build's of the " +
|
|
"catalogue, one key `list` (novox/hq ADR 0348 §7, design 57 K3): written by the controller alone from " +
|
|
"its store when a newer list is kept, read through `reads`"},
|
|
{DeclaredPackagesBucket, "each package the modules assigned to a node declare, present or absent, with the " +
|
|
"declaring modules, keyed by the node (novox/hq ADR 0507 §5): written by the controller alone when it " +
|
|
"composes the node, only on a change, read through `reads`"},
|
|
} {
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: m.bucket,
|
|
Description: m.why,
|
|
History: 1,
|
|
MaxValueSize: MirrorValueBytes,
|
|
MaxBytes: MirrorBytes,
|
|
Storage: jetstream.FileStorage,
|
|
LimitMarkerTTL: MirrorMarkerTTL,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", m.bucket, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// chainBucket is the options of a bucket of the chain: sixty-four revisions of each key, no age, a value up to
|
|
// ChainValueBytes, on disk.
|
|
func chainBucket(bucket, why string, bytes int64, markers time.Duration) jetstream.KeyValueConfig {
|
|
return jetstream.KeyValueConfig{
|
|
Bucket: bucket,
|
|
Description: why,
|
|
History: ChainHistory,
|
|
MaxValueSize: ChainValueBytes,
|
|
MaxBytes: bytes,
|
|
Storage: jetstream.FileStorage,
|
|
LimitMarkerTTL: markers,
|
|
}
|
|
}
|
|
|
|
// EnsureLeaseBucket creates the lease's bucket if absent and brings its options to match (to-be 45
|
|
// §6). **Before the lease is taken, by any candidate**: it is the lease's own precondition, and asserting
|
|
// a bucket that exists changes nothing. File storage, so its revisions — the epochs — outlive a restart of
|
|
// the bus; one raised again from nothing is moved past the highest epoch issued when the lease is taken.
|
|
func EnsureLeaseBucket(ctx context.Context, js jetstream.JetStream) error {
|
|
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
|
Bucket: LeaseBucket,
|
|
Description: "the controller's lease (novox/hq to-be 45 §6): the key `holder`, written by compare-and-set " +
|
|
"by the one controller instance that may act and renewed every five seconds; its revision when taken " +
|
|
"is the epoch every declaration and plan write carries",
|
|
History: 1,
|
|
TTL: LeaseTTL,
|
|
MaxValueSize: 4 << 10,
|
|
MaxBytes: 1 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}); err != nil {
|
|
return fmt.Errorf("asserting bucket %s: %w", LeaseBucket, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// HaveBucketConfig is the options of the have bucket (novox/hq ADR 0429 §1): delivery.HaveHistory revisions of each
|
|
// key, no age, 64 MiB on disk like the assignments. A module's have is JSON of about 500 bytes, but a node's own key
|
|
// carries its whole facts (its profile among them), several kilobytes on a desktop: a value is as large as the chain's
|
|
// buckets take, since a value is never cut (novox/hq ADR 0408 §6, issue 1162).
|
|
func HaveBucketConfig() jetstream.KeyValueConfig {
|
|
return jetstream.KeyValueConfig{
|
|
Bucket: delivery.HaveBucket,
|
|
Description: "what each machine has of each module, one key `<node>.<module>` (and `<node>.node` for the machine's " +
|
|
"own facts) (novox/hq ADR 0429, design 47): written by each machine's node-engine under its own node's keys " +
|
|
"alone, never by the controller; read by the delivery view and the watchman",
|
|
History: delivery.HaveHistory,
|
|
MaxValueSize: ChainValueBytes,
|
|
MaxBytes: 64 << 20,
|
|
Storage: jetstream.FileStorage,
|
|
}
|
|
}
|