AS13 (§181): the assignment graph store says filling, full and room; a full store keeps the last revision and writes no new want

Builds: design 47 G §181, §579 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 06:04:49 +01:00
parent de06949121
commit 84db958110
6 changed files with 222 additions and 1 deletions
@@ -0,0 +1,82 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
+
+	"mesh-delivery-assigner/internal/assigngraph"
+)
+
+// A store filling is said before it is full (design 47 G §181, in the assigner's part): the assignment graph revisions' object store
+// says `filling` of the seat mesh-wants when it passes 80 % of its cap, and `full` when the next revision does not fit: the assigner then keeps
+// its last revision and writes no new want, and says `room` when the store is back below 80 % and goes on. Nothing is dropped to make room.
+
+// fillOf is the store's fill now: its bytes and cap, when the bucket says them and has a cap.
+func (g *graphKeeper) fillOf(ctx context.Context) (used, capacity int64, ok bool) {
+	f, is := g.objects.(assigngraph.Filler)
+	if !is {
+		return 0, 0, false
+	}
+	used, capacity, err := f.Fill(ctx)
+	return used, capacity, err == nil && capacity > 0
+}
+
+func (p *passer) sayFill(event, about string, used, capacity int64, refused string) {
+	at := p.clock().UTC()
+	body := meshwants.Fill{Store: assigngraph.RevisionsObjects, Bytes: used, Cap: capacity, Refused: refused, At: &at}
+	if err := p.wantsSay(event, body, event+"-"+about); err != nil {
+		p.logf("%s of %s is not said: %v", event, assigngraph.RevisionsObjects, err)
+	}
+}
+
+// fillAtStart judges the store once at the process's first pass, when no event is pending: a store past 80 % says filling (full when at
+// its cap), so a fill reached while the process was down is said too.
+func (p *passer) fillAtStart(ctx context.Context) {
+	g := p.graph
+	used, capacity, ok := g.fillOf(ctx)
+	if !ok {
+		return
+	}
+	switch {
+	case used >= capacity:
+		g.full = true
+		p.sayFill(meshwants.EventFull, "start", used, capacity, "the next revision")
+	case used*5 >= capacity*4:
+		p.sayFill(meshwants.EventFilling, "start", used, capacity, "")
+	}
+}
+
+// fullBefore says the next revision does not fit its store, and keeps nothing: it says `full` once, and `room` once the store is below
+// 80 % again. It answers true while the store is full.
+func (p *passer) fullBefore(ctx context.Context, c assigngraph.Computed) bool {
+	g := p.graph
+	used, capacity, ok := g.fillOf(ctx)
+	if !ok {
+		return false
+	}
+	if g.full && used*5 < capacity*4 {
+		g.full = false
+		p.sayFill(meshwants.EventRoom, fmt.Sprint(used), used, capacity, "")
+	}
+	if used+assigngraph.NewBytes(ctx, g.objects, c) <= capacity {
+		g.full = false
+		return false
+	}
+	if !g.full {
+		g.full = true
+		p.sayFill(meshwants.EventFull, fmt.Sprint(used), used, capacity, "the next revision")
+	}
+	p.logf("the assignment graph revision is not kept: %s is full (%d of %d bytes), the last revision stays and no want is written until there is room", assigngraph.RevisionsObjects, used, capacity)
+	return true
+}
+
+// fillingAfter says `filling` when keeping a revision took the store past 80 % of its cap.
+func (p *passer) fillingAfter(ctx context.Context, before int64) {
+	g := p.graph
+	used, capacity, ok := g.fillOf(ctx)
+	if ok && before*5 < capacity*4 && used*5 >= capacity*4 {
+		p.sayFill(meshwants.EventFilling, fmt.Sprint(used), used, capacity, "")
+	}
+}
@@ -0,0 +1,83 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
+	meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
+
+	"mesh-delivery-assigner/internal/assigngraph"
+)
+
+// filled is an object bucket with a fill: the bytes it holds count beside what is stored.
+type filled struct {
+	objs map[string][]byte
+	base int64
+	cap  int64
+}
+
+func (f *filled) Put(_ context.Context, name string, data []byte) error {
+	f.objs[name] = data
+	return nil
+}
+func (f *filled) Get(_ context.Context, name string) ([]byte, error) {
+	if d, ok := f.objs[name]; ok {
+		return d, nil
+	}
+	return nil, assigngraph.ErrNoObject
+}
+func (f *filled) Names(context.Context) ([]string, error) {
+	out := []string{}
+	for n := range f.objs {
+		out = append(out, n)
+	}
+	return out, nil
+}
+func (f *filled) Fill(context.Context) (int64, int64, error) {
+	used := f.base
+	for _, d := range f.objs {
+		used += int64(len(d))
+	}
+	return used, f.cap, nil
+}
+
+// Design 47 G §181 (AS13): the assignment graph revisions' store says filling past 80 % of its cap, refuses a revision that does not fit
+// with `full` (the last revision stays, no want is written), and says `room` below 80 % again; at the start a store already past 80 % is said.
+func TestTheAssignmentGraphStoreSaysFillingFullAndRoom(t *testing.T) {
+	ctx := context.Background()
+	p := newPasser(kvSet{}, time.Now())
+	var events []string
+	p.publishWants = func(event string, body any, id string) error {
+		if _, ok := body.(meshwants.Fill); ok {
+			events = append(events, event)
+		}
+		return nil
+	}
+	o := &filled{objs: map[string][]byte{}, base: 500, cap: 1000}
+	p.graph = &graphKeeper{objects: o, halves: func() map[string]mcc.Half { return nil }}
+	p.keepGraph(ctx) // the first pass judges at the start: 500 of 1000 is below 80 %
+	if len(events) != 0 {
+		t.Fatalf("events %v", events)
+	}
+	o.base = 700
+	p.graph.started = false
+	p.keepGraph(ctx)
+	if len(events) != 1 || events[0] != meshwants.EventFilling {
+		t.Fatalf("at the start past 80 %%: %v", events)
+	}
+	c := assigngraph.Computed{Revision: meshwants.AssignmentRevision{Entries: []meshwants.NodeEntryRef{}}}
+	o.base = 990
+	if !p.fullBefore(ctx, c) || !p.graph.full || events[len(events)-1] != meshwants.EventFull {
+		t.Fatalf("a revision that does not fit: events %v", events)
+	}
+	p.fullBefore(ctx, c)
+	if n := len(events); n != 2 {
+		t.Errorf("full is said once: %v", events)
+	}
+	o.base = 100
+	if p.fullBefore(ctx, c) || p.graph.full || events[len(events)-1] != meshwants.EventRoom {
+		t.Errorf("room: events %v", events)
+	}
+}
@@ -36,6 +36,8 @@ type graphKeeper struct {
 	said   string
 	// cycled is the cycles last said (assigngraph.Cycles).
 	cycled string
+	// full says the next revision did not fit the store, so no want is written (graphfill.go); started says the first pass judged the fill.
+	full, started bool
 	// pins removes a revision that ceased to be current once nothing pins it (pins.go); nil keeps every revision.
 	pins *pinKeeper
 	// computed is the revision computed by the last pass, with its entries: the edges the wants wait on (actingwaits.go).
@@ -68,9 +70,17 @@ func (p *passer) keepGraph(ctx context.Context) {
 			p.logf("the assignment graph holds a cycle among the waits within one step, and its entries wait for each other: %s", cycles)
 		}
 	}
+	if !g.started {
+		g.started = true
+		p.fillAtStart(ctx)
+	}
 	if g.has && g.ok && assigngraph.Same(g.last, c.Revision) {
 		return
 	}
+	used, _, _ := g.fillOf(ctx)
+	if p.fullBefore(ctx, c) {
+		return
+	}
 	rev, fresh, err := assigngraph.Keep(ctx, g.objects, c, meshwants.EventAssignmentRevision, p.clock())
 	if err != nil {
 		p.logf("the assignment graph revision is not kept, so no want is said as wanted until it is: %v", err)
@@ -86,6 +96,7 @@ func (p *passer) keepGraph(ctx context.Context) {
 	if !fresh {
 		return
 	}
+	p.fillingAfter(ctx, used)
 	if g.pins != nil {
 		g.pins.judge() // the revision before ceased to be current
 	}
@@ -99,3 +99,20 @@ func (b *busObjects) Delete(ctx context.Context, name string) error {
 	}
 	return nil
 }
+
+// Fill is the bytes the bucket holds and its cap (assigngraph.Filler).
+func (b *busObjects) Fill(ctx context.Context) (int64, int64, error) {
+	o, err := b.open(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	st, err := o.Status(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	info, err := b.js.Stream(ctx, "OBJ_"+st.Bucket())
+	if err != nil {
+		return int64(st.Size()), 0, err
+	}
+	return int64(st.Size()), info.CachedInfo().Config.MaxBytes, nil
+}
@@ -205,7 +205,9 @@ func (p *passer) clock() time.Time {
 func (p *passer) pass(ctx context.Context) {
 	p.assignmentsSeat(ctx)
 	p.keepGraph(ctx)
-	p.wants(ctx)
+	if p.graph == nil || !p.graph.full { // a full assignment graph store keeps the last revision and writes no new want (design 47 G §181)
+		p.wants(ctx)
+	}
 	p.puts(ctx)
 	p.nodeWants(ctx)
 	p.backups(ctx)
@@ -149,3 +149,29 @@ func Event(rev meshwants.AssignmentRevision) meshwants.AssignmentRevisionEvent {
 	return meshwants.AssignmentRevisionEvent{V: rev.V, Revision: rev.Revision, Kind: rev.Kind, Cause: rev.Cause, Group: rev.Group,
 		Heads: rev.Heads, Catalogs: rev.Catalogs, At: rev.At}
 }
+
+// Filler is an object bucket that says its fill: the bytes used and its cap in bytes (0: no cap known).
+type Filler interface {
+	Fill(ctx context.Context) (used, capacity int64, err error)
+}
+
+// NewBytes is what keeping c next would add to the bucket: each entry object not held yet and the list (design 47 G §181).
+func NewBytes(ctx context.Context, o Objects, c Computed) int64 {
+	names, err := o.Names(ctx)
+	if err != nil {
+		return 0
+	}
+	held := make(map[string]bool, len(names))
+	for _, n := range names {
+		held[n] = true
+	}
+	var size int64
+	for _, ref := range c.Revision.Entries {
+		if e, ok := c.Entries[ref.Digest]; ok && !held[EntryObject(ref.Digest)] {
+			data, _ := json.Marshal(e)
+			size += int64(len(data))
+		}
+	}
+	list, _ := json.Marshal(c.Revision)
+	return size + int64(len(list))
+}