A-370: the verb registrations is answered from its state at once, not queued behind a take-in
Builds: design 47 G A-370 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -0,0 +1,60 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
|
||||
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
type noneWiring struct{ followers []*views.Follower }
|
||||
|
||||
func (w *noneWiring) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error {
|
||||
w.followers = append(w.followers, f)
|
||||
return nil
|
||||
}
|
||||
func (w *noneWiring) Drop(string, string) error { return nil }
|
||||
|
||||
// Design 47 G A-372: the assigner's wait for its catalog half asks the catalog client's Ready as well: a mesh whose registrations and
|
||||
// gate's source are answered and name no branch is none-yet, and the first pass is told so once; before they answer it is not told.
|
||||
func TestAMeshThatNamesNoBranchSettlesTheAssignersFirstPass(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
call := func(key string, _ any) (json.RawMessage, error) {
|
||||
if key == mcc.RegistrationsVerb {
|
||||
return json.Marshal(seat.RegistrationsAnswer{Registrations: []seat.Registration{}})
|
||||
}
|
||||
return json.Marshal(map[string][]string{"branches": {}})
|
||||
}
|
||||
for _, answered := range []bool{false, true} {
|
||||
w := &noneWiring{}
|
||||
b, err := mcc.NewBranches(w, call)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
settled := make(chan struct{}, 2)
|
||||
h := &halves{ctx: ctx, settled: settled, branches: b}
|
||||
h.settleWhenNoBranch()
|
||||
if answered {
|
||||
for _, f := range w.followers {
|
||||
if err := f.Start(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-settled:
|
||||
case <-ctx.Done():
|
||||
t.Fatal("a mesh that names no branch never settled the first pass")
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-settled:
|
||||
t.Errorf("answered %v: the first pass was told twice or before the first views", answered)
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -48,6 +48,12 @@ func (b *Branches) namesNoBranch() bool {
|
||||
default:
|
||||
return false
|
||||
}
|
||||
// A first view closes before the follower's change is taken into the branches followed, so what the views hold is read too.
|
||||
for _, reg := range b.Registrations.All() {
|
||||
if len(reg.Branches) > 0 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
b.mu.Lock()
|
||||
s := b.source
|
||||
b.mu.Unlock()
|
||||
@@ -56,10 +62,35 @@ func (b *Branches) namesNoBranch() bool {
|
||||
}
|
||||
select {
|
||||
case <-s.Follower.Ready():
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
return len(s.Branches()) == 0
|
||||
}
|
||||
|
||||
// FirstViews is closed once the first views of the registrations and, while it is followed, of the gate's source are in: a part
|
||||
// that waits for the catalog half to settle looks at Followed and Ready once it is closed (design 47 G, A-372).
|
||||
func (b *Branches) FirstViews(ctx context.Context) <-chan struct{} {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
select {
|
||||
case <-b.Registrations.Ready():
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
b.mu.Lock()
|
||||
s := b.source
|
||||
b.mu.Unlock()
|
||||
if s != nil {
|
||||
select {
|
||||
case <-s.Follower.Ready():
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
close(done)
|
||||
}()
|
||||
return done
|
||||
}
|
||||
|
||||
// Entry is a module's entry at the trunk: the module's own registration decides, so the branch followed first, which lists
|
||||
|
||||
@@ -310,7 +310,7 @@ func main() {
|
||||
return err
|
||||
})...)
|
||||
tools = append(tools, index.ConsumerTools(keptRead{stdio.State(process.ConsumersState)})...)
|
||||
tools = append(tools, index.RegistrationTools(keptRead{stdio.State(index.ProposalsState)}, asks, time.Now, func(cause, verb, what, why string) error {
|
||||
tools = append(tools, index.RegistrationTools(keptRead{stdio.State(index.RegistrationsState)}, keptRead{stdio.State(index.ProposalsState)}, asks, time.Now, func(cause, verb, what, why string) error {
|
||||
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause, Verb: verb})
|
||||
return err
|
||||
})...)
|
||||
|
||||
@@ -15,7 +15,7 @@ import (
|
||||
// the indexer's own tools of the proposals (§667) and the seed (§8), served in the tools bundle. Every act is asked of the
|
||||
// indexer through takeins, as the indexer's other tools are (ask writes the value and waits for the answer); the proposals
|
||||
// are read from the state as the bundle reads it. record says a hand act in the hand-act log.
|
||||
func RegistrationTools(proposals Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(cause, verb, what, why string) error) []stdio.Tool {
|
||||
func RegistrationTools(registrations, proposals Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(cause, verb, what, why string) error) []stdio.Tool {
|
||||
act := func(verb string, withInput bool) func(map[string]any) (any, error) {
|
||||
return func(args map[string]any) (any, error) {
|
||||
var input json.RawMessage
|
||||
@@ -39,8 +39,31 @@ func RegistrationTools(proposals Store, ask func(key string, value []byte) (Take
|
||||
return answerOf(ask(key, value))
|
||||
}
|
||||
}
|
||||
// The verb registrations is a first view of the catalog half: it is answered at once from the state the registrations are kept
|
||||
// in, never queued behind a running take-in (design 47 G A-370). Only a module's history is read from the stream the
|
||||
// indexer keeps, so that is still asked of the indexer.
|
||||
viaIndexer := act(ActRegistrations, true)
|
||||
return []stdio.Tool{
|
||||
{Name: meshcatalog.Seat + "." + meshcatalog.VerbRegistrations, Run: act(ActRegistrations, true)},
|
||||
{Name: meshcatalog.Seat + "." + meshcatalog.VerbRegistrations, Run: func(args map[string]any) (any, error) {
|
||||
var in meshcatalog.RegistrationsInput
|
||||
clean := map[string]any{}
|
||||
for k, v := range args {
|
||||
if k != "caller" && !strings.HasPrefix(k, "_") {
|
||||
clean[k] = v
|
||||
}
|
||||
}
|
||||
raw, err := json.Marshal(clean)
|
||||
if err == nil {
|
||||
err = json.Unmarshal(raw, &in)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if in.History {
|
||||
return viaIndexer(args)
|
||||
}
|
||||
return (&Registry{Registrations: registrations}).RegistrationsAnswer(context.Background(), in)
|
||||
}},
|
||||
{Name: meshcatalog.Seat + "." + meshcatalog.VerbRegister, Run: act(verbRegister, true)},
|
||||
{Name: meshcatalog.Seat + "." + meshcatalog.VerbPolicy, Run: act(verbPolicy, true)},
|
||||
{Name: meshcatalog.Seat + "." + meshcatalog.VerbWatch, Run: act(verbWatch, true)},
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
)
|
||||
|
||||
// Design 47 G A-370: the verb registrations answers from the state at once; it never asks the indexer (whose take-in may be
|
||||
// running), but for a module's history, which the indexer reads from its stream.
|
||||
func TestTheRegistrationsVerbIsAnsweredFromItsStateNotAskedOfTheIndexer(t *testing.T) {
|
||||
store := MemoryStore()
|
||||
data, _ := json.Marshal(meshcatalog.Registration{Module: "a"})
|
||||
if err := store.Create(context.Background(), RegistrationKey("a"), data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
asked := 0
|
||||
ask := func(string, []byte) (TakeInAsk, error) { asked++; return TakeInAsk{}, errors.New("the take-in is still running") }
|
||||
var tool func(map[string]any) (any, error)
|
||||
for _, tl := range RegistrationTools(store, MemoryStore(), ask, time.Now, nil) {
|
||||
if tl.Name == meshcatalog.Seat+"."+meshcatalog.VerbRegistrations {
|
||||
tool = tl.Run
|
||||
}
|
||||
}
|
||||
for _, args := range []map[string]any{{}, {"module": "a"}} {
|
||||
got, err := tool(args)
|
||||
if err != nil || len(got.(meshcatalog.RegistrationsAnswer).Registrations) != 1 {
|
||||
t.Errorf("registrations %v = %+v, %v, want the one registration", args, got, err)
|
||||
}
|
||||
}
|
||||
if asked != 0 {
|
||||
t.Errorf("the indexer was asked %d times for a read of the state", asked)
|
||||
}
|
||||
if _, err := tool(map[string]any{"module": "a", "history": true}); err == nil || asked != 1 {
|
||||
t.Errorf("a history is asked of the indexer: err %v, asked %d", err, asked)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user