Merge branch 'feat/47-G-fix9-grants' into feat/47-G-join9
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
)
|
||||
|
||||
// The part's own membership names the stream running among those it hears, and hearRunning binds the part's durable on it:
|
||||
// the loop that opens the other streams leaves it alone, so no second reader takes messages of that one durable (design 47 G
|
||||
// §556, A-19; the whole-part integration test lost the running event to the second reader at random).
|
||||
func TestTheLoopOfOwnHearsLeavesTheStreamsADedicatedHearingOwns(t *testing.T) {
|
||||
hears := []membership.HeardIssued{{Stream: nodeengine.StreamRunning, Consumer: "a_mesh-grants"}, {Stream: "mod-c_feed", Consumer: "a_mesh-grants"}, {Stream: "opened_one"}}
|
||||
got := hearsToOpen(hears, map[string]bool{"opened_one": true})
|
||||
if len(got) != 1 || got[0].Stream != "mod-c_feed" {
|
||||
t.Errorf("the streams opened are %+v, want mod-c_feed alone", got)
|
||||
}
|
||||
}
|
||||
@@ -342,7 +342,7 @@ func TestTheGrantsPartWhole(t *testing.T) {
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("never saw: %s\n%s", what, logged.tail("binding"))
|
||||
t.Fatalf("never saw: %s\n%s\n%s", what, logged.tail("binding"), logged.tail("running")+logged.tail("events:")+logged.tail("cannot be heard"))
|
||||
}
|
||||
until("every first view asked and refused once", func() bool { return refusedAsks.Load() >= int32(len(grants.FirstViews)) })
|
||||
until("the part saying it waits on the indexer and the assigner", func() bool {
|
||||
@@ -573,7 +573,9 @@ func TestTheGrantsPartWhole(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for m := range got.Messages() {
|
||||
_ = m.Ack()
|
||||
if err := m.DoubleAck(ctx); err != nil { // the ack settled on the server before the consumer is read
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
before, _ := stream.Info(ctx)
|
||||
cbefore, _ := cons.Info(ctx)
|
||||
|
||||
@@ -41,6 +41,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
@@ -258,10 +259,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, h := range hears {
|
||||
if opened[h.Stream] {
|
||||
continue
|
||||
}
|
||||
for _, h := range hearsToOpen(hears, opened) {
|
||||
cons, err := js.Consumer(ctx, h.Stream, h.Consumer)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the consumer %s on %s: %w", h.Consumer, h.Stream, err)
|
||||
@@ -292,6 +290,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return nil
|
||||
}
|
||||
|
||||
// hearsToOpen are the streams of the part's own membership no hearing has opened yet. A stream another hearing of the part
|
||||
// binds its durable on (running: hearRunning, which reads it whole at the start before it follows) is left to that hearing:
|
||||
// two readers of one durable each take the messages the other does not, and the event a reader of its own never hears is lost
|
||||
// to it (the integration test of the whole part met it as a state running that stayed empty, 2026-11-12).
|
||||
func hearsToOpen(hears []membership.HeardIssued, opened map[string]bool) []membership.HeardIssued {
|
||||
var out []membership.HeardIssued
|
||||
for _, h := range hears {
|
||||
if !opened[h.Stream] && h.Stream != nodeengine.StreamRunning {
|
||||
out = append(out, h)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// hearing is the stream the part's running consumer is bound to, once.
|
||||
var hearingRunning struct {
|
||||
sync.Mutex
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package buscompose
|
||||
|
||||
import "testing"
|
||||
|
||||
// Design 47 G A-77, A-92: the node-engine's account is granted the state bounds, to write a bound key with its time to live,
|
||||
// to delete it, and to watch it for the bus's marker; and no other node-engine state or module's bucket by that grant.
|
||||
func TestTheNodeEngineIsGrantedItsBoundsState(t *testing.T) {
|
||||
users, err := Users(Records{Nodes: []string{"one"}, Engine: &Declared{Module: EngineModule}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var node Principal
|
||||
for _, u := range users {
|
||||
if u.Kind == KindNode {
|
||||
node = u
|
||||
}
|
||||
}
|
||||
perms, err := PermissionsFor(node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, want := range []string{
|
||||
"$KV.mesh-host_bounds.recovery.act-1", "$JS.API.STREAM.INFO.KV_mesh-host_bounds",
|
||||
"$JS.API.CONSUMER.CREATE.KV_mesh-host_bounds.*.$KV.mesh-host_bounds.recovery.act-1",
|
||||
"$JS.API.CONSUMER.DELETE.KV_mesh-host_bounds.x",
|
||||
} {
|
||||
if !MayPublish(perms, want) {
|
||||
t.Errorf("the node-engine of one may not publish %s", want)
|
||||
}
|
||||
}
|
||||
if MayPublish(perms, "$KV.other_bounds.recovery.act-1") {
|
||||
t.Error("the node-engine of one may publish another module's bounds")
|
||||
}
|
||||
}
|
||||
@@ -34,6 +34,17 @@ var EnginePackagesBucket = BucketName(EngineModule, EnginePackagesState)
|
||||
// condition.
|
||||
var EngineFindingsBucket = BucketName(EngineModule, "findings")
|
||||
|
||||
// EngineBoundsBucket is the engine's state `bounds` (design 47 G §560, A-77, A-92): one key `<what>.<id>` for each bound a wait of a
|
||||
// node's engine runs against, written with its own time to live by that engine, watched for the bus's marker when it ends and
|
||||
// deleted when the wait ends first. The key names no node, so each node's engine is granted the whole bucket.
|
||||
var EngineBoundsBucket = BucketName(EngineModule, "bounds")
|
||||
|
||||
// EngineBoundsGrants is what a node's engine publishes to keep a bound: the writes of the bucket's keys (the create with its time to
|
||||
// live, the delete) and what a watch of them needs.
|
||||
func EngineBoundsGrants() []string {
|
||||
return append([]string{"$KV." + EngineBoundsBucket + ".>"}, readGrants(EngineBoundsBucket, "")...)
|
||||
}
|
||||
|
||||
// EngineFindingsGrants is what a node's engine publishes to write and clear its node's findings: its own node's keys alone,
|
||||
// and their read. Without it the finding that the node waits for the seat list is refused on the bus and never raised.
|
||||
func EngineFindingsGrants(node string) []string {
|
||||
@@ -126,5 +137,7 @@ func engineGrants(p Principal) (pub, sub []string) {
|
||||
if safeSubject.MatchString(p.Node) {
|
||||
pub = append(pub, EngineFindingsGrants(p.Node)...)
|
||||
}
|
||||
// And the bounds of its waits (design 47 G A-77, A-92): the state bounds, written with a time to live and watched.
|
||||
pub = append(pub, EngineBoundsGrants()...)
|
||||
return append(pub, tp...), append(sub, ts...)
|
||||
}
|
||||
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -213,6 +213,12 @@ var WritersTable = []WriterRow{
|
||||
{State: "a machine's data items", Writer: "the node-engine of the machine, under its own keys `<node>.>`",
|
||||
KeptIn: "key-value " + EngineDataBucket, Others: "read by the node's disk-load through `reads`, its own node's keys alone",
|
||||
Subjects: kvOf(EngineDataBucket), Writes: ownNodeKeys},
|
||||
// The bounds of a node's engine's waits (design 47 G A-77, A-92): keys `<what>.<id>` with a time to live of their own, which name no
|
||||
// node, so every node's engine writes the bucket.
|
||||
{State: "a node-engine's bounds", Writer: "the node-engine of any machine, keys `<what>.<id>` with a time to live",
|
||||
KeptIn: "key-value " + EngineBoundsBucket + ", the node-engine's declared state bounds",
|
||||
Others: "read by nobody but the engines: the bus's marker of an expired key ends the wait", Subjects: kvOf(EngineBoundsBucket),
|
||||
Writes: func(p Principal, pattern string) bool { return p.Kind == KindNode }},
|
||||
{State: "a machine's hand acts", Writer: "the node-engine of the machine, under its own keys `<node>.>`, compare-and-set",
|
||||
KeptIn: "key-value " + EngineActsBucket + ", the node-engine's declared state acts",
|
||||
Others: "read through the node seat's verb acts", Subjects: kvOf(EngineActsBucket),
|
||||
|
||||
Reference in New Issue
Block a user