mesh-delivery-assigner: read grants from the source the lease names (issue 774, ADR 0547 §4)
Why: from the lease write the grants part's state grants is the source and the controller's bucket otherwise; the assigner follows the lease key (a pass only when the source changes), and neither bucket stops its start. Needs the group's mesh-sdk pin (GrantSource, LeaseHandedAway, the assigner a named reader of the lease).
This commit is contained in:
@@ -177,8 +177,8 @@ func saysWhatItKeeps(t *testing.T, b *bed, done <-chan error) {
|
||||
// 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket.
|
||||
func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
|
||||
for _, missing := range readBuckets {
|
||||
if missing == delivery.PlacementBucket {
|
||||
continue // it starts without it, below
|
||||
if missing == delivery.PlacementBucket || sourceOfGrants(missing) {
|
||||
continue // it starts without it, below (the grants' source: main_test.go)
|
||||
}
|
||||
b := startBed(t, missing)
|
||||
select {
|
||||
|
||||
@@ -54,6 +54,8 @@ var readBuckets = []string{
|
||||
provider.AssignmentsBucket,
|
||||
delivery.BuildsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
delivery.PartGrantsBucket,
|
||||
buscompose.LeaseBucket,
|
||||
delivery.ControllerComposedBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
@@ -117,6 +119,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
||||
continue
|
||||
}
|
||||
if sourceOfGrants(name) && err != nil {
|
||||
// The grants part's grants and the controller's lease choose where grants are read from (issue 774): without
|
||||
// them the controller's grants are read, as before the lease is handed away.
|
||||
logf("the bucket %s cannot be opened (%v): the controller's grants are read until the assigner starts again", name, err)
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err)
|
||||
}
|
||||
@@ -125,6 +133,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// sourceOfGrants says the bucket is one of the two that choose the grants' source, the grants part's grants and the
|
||||
// controller's lease: neither stops the assigner when it is not there (delivery.GrantSource).
|
||||
func sourceOfGrants(name string) bool {
|
||||
return name == delivery.PartGrantsBucket || name == buscompose.LeaseBucket
|
||||
}
|
||||
|
||||
// run is the program: it returns when ctx ends or the bus cannot be used.
|
||||
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
|
||||
logger := log.New(logOut, "mesh-delivery-assigner: ", log.LstdFlags)
|
||||
@@ -170,6 +184,8 @@ var watched = []string{
|
||||
buscompose.ResolvedBucket,
|
||||
provider.AssignmentsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
delivery.PartGrantsBucket,
|
||||
buscompose.LeaseBucket,
|
||||
delivery.ControllerComposedBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
@@ -233,7 +249,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
value = e.Value()
|
||||
}
|
||||
before := w.grantsBucket()
|
||||
w.putEntry(name, e.Key(), value, e.Revision(), e.Created())
|
||||
if name == buscompose.LeaseBucket && w.grantsBucket() == before {
|
||||
// The controller renews its lease every few seconds: only a change of where grants are read from
|
||||
// makes a pass (delivery.GrantSource).
|
||||
continue
|
||||
}
|
||||
wake()
|
||||
}
|
||||
}
|
||||
@@ -275,7 +297,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
started := 0
|
||||
for _, name := range watched {
|
||||
if kvs[name] == nil && slices.Contains(seatBuckets, name) {
|
||||
if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name)) {
|
||||
continue // not on the bus at this start (open said so)
|
||||
}
|
||||
started++
|
||||
@@ -297,13 +319,21 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
name string
|
||||
kv jetstream.KeyValue
|
||||
}
|
||||
placementAppeared := make(chan appeared, len(seatBuckets))
|
||||
placementAppeared := make(chan appeared, len(seatBuckets)+2)
|
||||
for _, name := range seatBuckets {
|
||||
if kvs[name] == nil && js != nil {
|
||||
process.AwaitBucket(ctx, js, nc, name, "the seat mesh-placement waits for it",
|
||||
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- appeared{name, kv} })
|
||||
}
|
||||
}
|
||||
// And the two that choose where grants are read from, with the same delivery as the grants part that writes one.
|
||||
for _, name := range []string{delivery.PartGrantsBucket, buscompose.LeaseBucket} {
|
||||
if kvs[name] == nil && js != nil {
|
||||
name := name
|
||||
process.AwaitBucket(ctx, js, nc, name, "the controller's grants are read",
|
||||
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- appeared{name, kv} })
|
||||
}
|
||||
}
|
||||
takePlacement := func(a appeared) error {
|
||||
if kvs[a.name] != nil {
|
||||
return nil // bound already, by the other of its two events
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
@@ -70,12 +71,15 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
{delivery.AssignerPutsBucket, jetstream.ErrBucketNotFound, true},
|
||||
{delivery.AssignerPutsBucket, errors.New("nats: permissions violation for subscription to \"$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_puts\""), true},
|
||||
{delivery.AssignerPutsBucket, context.DeadlineExceeded, true},
|
||||
// Issue 774: the grants part's grants and the controller's lease choose the grants' source; neither stops the start.
|
||||
{delivery.PartGrantsBucket, jetstream.ErrBucketNotFound, true},
|
||||
{buscompose.LeaseBucket, errors.New("nats: permissions violation"), true},
|
||||
{delivery.AssignerWantsBucket, errors.New("nats: permissions violation"), false},
|
||||
} {
|
||||
var logged string
|
||||
kvs, err := open(context.Background(), refusingJS{missing: c.missing, err: c.err}, func(f string, a ...any) { logged = fmt.Sprintf(f, a...) })
|
||||
switch {
|
||||
case c.starts && (err != nil || kvs[c.missing] != nil || len(kvs) != len(readBuckets)+len(ownBuckets)-1 || !strings.Contains(logged, "no put is taken further")):
|
||||
case c.starts && (err != nil || kvs[c.missing] != nil || len(kvs) != len(readBuckets)+len(ownBuckets)-1 || !strings.Contains(logged, "until the assigner starts again")):
|
||||
t.Errorf("%s refused with %v: the start answered %v, %d buckets, logged %q", c.missing, c.err, err, len(kvs), logged)
|
||||
case !c.starts && (err == nil || !strings.Contains(err.Error(), "grant of "+c.missing)): // 1945: named with its grant
|
||||
t.Errorf("%s refused with %v: the start answered %v", c.missing, c.err, err)
|
||||
|
||||
@@ -146,6 +146,8 @@ func (p *passer) wants(ctx context.Context) {
|
||||
stale := map[string]neitherOne{}
|
||||
left := map[string]string{}
|
||||
defer func() { p.sayNeither(neither, stale, left) }()
|
||||
// Where grants are read from, once for the pass: the grants part's while the lease is handed away (issue 774).
|
||||
grants := p.w.grantsBucket()
|
||||
for name := range names {
|
||||
fp := trunk[name].FP2
|
||||
name := name
|
||||
@@ -161,7 +163,7 @@ func (p *passer) wants(ctx context.Context) {
|
||||
},
|
||||
GrantedAt: func(fp string) bool {
|
||||
key, err := delivery.ControllerGrantKey(name, fp)
|
||||
return err == nil && p.w.has(delivery.ControllerGrantsBucket, key)
|
||||
return err == nil && p.w.has(grants, key)
|
||||
},
|
||||
Restored: func(node string) bool {
|
||||
key, err := delivery.RestoreAskKey(node, name, fp)
|
||||
@@ -177,8 +179,8 @@ func (p *passer) wants(ctx context.Context) {
|
||||
hold, _ := delivery.HoldAskKey(name, fp)
|
||||
m.Released = p.answeredWith(hold, "release")
|
||||
m.Kept = p.answeredWith(hold, "keep")
|
||||
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held) && !m.Released, p.w.has(delivery.ControllerGrantsBucket, grant)
|
||||
if raw := p.w.value(delivery.ControllerGrantsBucket, grant); raw != nil {
|
||||
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held) && !m.Released, p.w.has(grants, grant)
|
||||
if raw := p.w.value(grants, grant); raw != nil {
|
||||
if g, err := delivery.DecodeControllerGrant(raw); err == nil {
|
||||
m.GrantedOn = func(node string) bool { return assign.GrantNamesNode(g, node, name) }
|
||||
}
|
||||
|
||||
@@ -213,8 +213,7 @@ func (p *passer) putWorld(node, module string, put delivery.Put, manifest catalo
|
||||
|
||||
// composedFor is the composition of module for node at fp and whether the grant at fp names the node's accounts: the one
|
||||
// place a put reads them (ADR 0546 clause 4). Today their writer is the controller (mesh-controller_composed and
|
||||
// mesh-controller_grants); when composing and grants move out of the controller into delivery's own parts, this reads them
|
||||
// from their new owner and nothing else changes.
|
||||
// mesh-controller_grants); the grants are the grants part's while the controller's lease is handed away (grantsBucket).
|
||||
func (p *passer) composedFor(node, module, fp string) (*delivery.ComposedDigest, bool) {
|
||||
var out *delivery.ComposedDigest
|
||||
if d, ok := p.w.composed(node, module, fp); ok {
|
||||
@@ -222,7 +221,7 @@ func (p *passer) composedFor(node, module, fp string) (*delivery.ComposedDigest,
|
||||
}
|
||||
granted := false
|
||||
if gkey, err := delivery.ControllerGrantKey(module, fp); err == nil {
|
||||
if raw := p.w.value(delivery.ControllerGrantsBucket, gkey); raw != nil {
|
||||
if raw := p.w.value(p.w.grantsBucket(), gkey); raw != nil {
|
||||
if g, err := delivery.DecodeControllerGrant(raw); err == nil {
|
||||
granted = g.NamesNode(node, module)
|
||||
}
|
||||
|
||||
@@ -546,6 +546,16 @@ func (w *world) value(bucket, key string) []byte {
|
||||
return w.values[bucket][key]
|
||||
}
|
||||
|
||||
// grantsBucket is the bucket the grants are read from now (delivery.GrantSource): the grants part's while the controller's
|
||||
// lease is handed away, as the lease bucket's key says it (buscompose.LeaseHandedAway), the controller's in every other
|
||||
// case, a lease not held here included.
|
||||
func (w *world) grantsBucket() string {
|
||||
raw := w.value(buscompose.LeaseBucket, buscompose.ControllerLeaseKey)
|
||||
away, _ := buscompose.LeaseHandedAway(raw, raw != nil, nil)
|
||||
bucket, _ := delivery.GrantSource(away)
|
||||
return bucket
|
||||
}
|
||||
|
||||
// has says whether a bucket holds a key now.
|
||||
func (w *world) has(bucket, key string) bool {
|
||||
w.mu.Lock()
|
||||
|
||||
@@ -6,7 +6,9 @@ import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
@@ -125,4 +127,40 @@ func theWorldCountsTheValuesAKeyTakes(t *testing.T) {
|
||||
if got := w.changed(delivery.HaveBucket, "gone"); got != 0 {
|
||||
t.Errorf("changed = %d for a key never seen", got)
|
||||
}
|
||||
// Issue 774: grants are read from the grants part's bucket only while the controller's lease is handed away: a grant
|
||||
// only the part holds lets a node's put through then, and with a controller's lease (or none, or an unreadable one) it
|
||||
// holds it, a grant only the controller holds being the one that counts instead.
|
||||
fp := delivery.SourcePrefix2 + strings.Repeat("d", 64)
|
||||
gkey, _ := delivery.ControllerGrantKey("web", fp)
|
||||
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{Accounts: []string{"ace.web"}})
|
||||
for _, row := range []struct {
|
||||
name, lease string
|
||||
inPart, inController bool
|
||||
bucket string
|
||||
granted bool
|
||||
}{
|
||||
{"handed away, the part's grant", `{"handed-to":"the operator's hand-over"}`, true, false, delivery.PartGrantsBucket, true},
|
||||
{"handed away, only the controller's grant", `{"handed-to":"x"}`, false, true, delivery.PartGrantsBucket, false},
|
||||
{"a controller holds it, only the part's grant", `{"instance":"novox/1","epoch":3}`, true, false, delivery.ControllerGrantsBucket, false},
|
||||
{"a controller holds it, its own grant", `{"instance":"novox/1"}`, false, true, delivery.ControllerGrantsBucket, true},
|
||||
{"no lease read", ``, true, false, delivery.ControllerGrantsBucket, false},
|
||||
{"an unreadable lease", `[1]`, true, false, delivery.ControllerGrantsBucket, false},
|
||||
} {
|
||||
p := newPasser(buckets{}, time.Now())
|
||||
if row.lease != "" {
|
||||
p.w.put(buscompose.LeaseBucket, buscompose.ControllerLeaseKey, []byte(row.lease))
|
||||
}
|
||||
if row.inPart {
|
||||
p.w.put(delivery.PartGrantsBucket, gkey, grant)
|
||||
}
|
||||
if row.inController {
|
||||
p.w.put(delivery.ControllerGrantsBucket, gkey, grant)
|
||||
}
|
||||
if got := p.w.grantsBucket(); got != row.bucket {
|
||||
t.Errorf("%s: grants are read from %s, want %s", row.name, got, row.bucket)
|
||||
}
|
||||
if _, granted := p.composedFor("ace", "web", fp); granted != row.granted {
|
||||
t.Errorf("%s: granted = %v, want %v", row.name, granted, row.granted)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,12 +24,14 @@
|
||||
"mesh-controller.builds",
|
||||
"mesh-controller.composed",
|
||||
"mesh-controller.grants",
|
||||
"mesh-controller.lease",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-teller.answers",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"mesh-grants.grants",
|
||||
"node-engine.have"
|
||||
],
|
||||
"uses": [
|
||||
|
||||
Reference in New Issue
Block a user