From 8ee28d08c33c8e8e3fe834a4a1205af0f8016954 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 14:44:17 +0100 Subject: [PATCH] The gate says its records' ledger filling, full and room, and holds a merge whose record finds no room (milestone 47/G, W11.20) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Design 47: a store filling is said before it is full. The gate reads its ledger's fill through the tool runner's mesh/ledger.fill, says filling of the seat mesh-delivery when a record takes the ledger past 80 % of its cap, refuses the next record when none fits and says full naming it, and keeps the merge waiting for room on its delivery (held_full), so a restarted gate judges alike. Once the ledger is below 80 % again, after the retention's removals, it says room and hands the merge. gate_records says the fill against its cap, or why it is unread. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ (Joined from 061be00d.) Builds: design 47 G §181 (GATE12) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-gate/main.go | 2 +- .../cmd/mesh-delivery-gate/merges.go | 8 + .../cmd/mesh-delivery-gate/ready.go | 5 +- .../cmd/mesh-delivery-gate/record.go | 12 +- .../cmd/mesh-delivery-gate/record_test.go | 125 +++++++++++ .../cmd/mesh-delivery-gate/recordfill.go | 201 ++++++++++++++++++ .../cmd/mesh-delivery-gate/recordtools.go | 6 +- .../cmd/mesh-delivery-gate/switch.go | 29 +-- 8 files changed, 369 insertions(+), 19 deletions(-) create mode 100644 modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordfill.go diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index 468c75217..bf165028f 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -74,7 +74,7 @@ func main() { } h.init() h.SaySeat = stdio.SeatPublish - h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention, Say: stdio.SeatPublish} + h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention, Say: stdio.SeatPublish, Logf: logf} h.Merges = recordedMerges{MergeStore: kvMerges{}, records: h.Records, members: h.groupMembers} h.Old = kvOld{} h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go index 3f32bfb77..e9701a0af 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go @@ -291,6 +291,9 @@ func (h *Holder) hand(gid, name, key, target string, reads []memberRead, approve if gid != "" && !h.groupMergeable(gid, target, reads) { return } + if h.waitForRoom(target) { + return + } h.mu.Lock() value, ok := h.mergeValue(name, key, approved, target, mdc.MergeMerging) h.mu.Unlock() @@ -306,6 +309,11 @@ func (h *Holder) hand(gid, name, key, target string, reads []memberRead, approve err = h.Merges.Put(name, value) } if err != nil { + if errors.Is(err, ErrRecordsFull) { + h.mu.Lock() + h.holdFull(target, h.Now()) + h.mu.Unlock() + } h.giveBack(target, err) return } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go index f3da1acf1..649f99a27 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go @@ -136,7 +136,10 @@ type ReadyRun struct { // CannotWrite is why the operator is not asked: the forge's merging account cannot write a repository the ask covers // (writes.go, issue 1945); CannotWriteSince since when. Said at once as a condition. - CannotWrite string `json:"cannot_write,omitempty"` + CannotWrite string `json:"cannot_write,omitempty"` + // HeldFull is since when its merge waits for room in the delivery records' ledger, which refused its record as full + // (recordfill.go): handed once the ledger is below 80 % of its cap again. + HeldFull *time.Time `json:"held_full,omitempty"` CannotWriteSince *time.Time `json:"cannot_write_since,omitempty"` // NotHanded is why an approved merge that waits for nothing is not handed to the forge bridge (a member of its group that diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go index 9cd40c3ce..38b81225a 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go @@ -345,8 +345,11 @@ type Records struct { Now func() time.Time // Retention is how long a record is kept; nil keeps DefaultRecordRetention. Retention func() time.Duration - // Say says the seat mesh-delivery's events of the records: pinned, freed, record-ended (recordevents.go); nil says none. + // Say says the seat mesh-delivery's events of the records: pinned, freed, record-ended (recordevents.go), filling, + // full and room (recordfill.go); nil says none. Say SeatSay + // Logf says what the records could not say; nothing is said when nil. + Logf func(format string, args ...any) // held are the records kept, by instance, read once from the ledger (Current) and kept as the gate writes them: the // gate is their one writer. @@ -475,6 +478,10 @@ func (rs *Records) Open(milestone string, heads map[string]string, forgeMade boo if err := rs.Sweep(now); err != nil { return nil, false, fmt.Errorf("the records past their retention cannot be removed, so no record is opened: %w", err) } + before, read, err := rs.refuseFull(milestone, inst) + if err != nil { + return nil, false, err + } body := map[string]any{"milestone": milestone, "heads": heads, "opened": now, "form": RecordForm} if forgeMade { body["forge-made"] = true @@ -485,6 +492,9 @@ func (rs *Records) Open(milestone string, heads map[string]string, forgeMade boo if _, err := rs.Ledger.Apply(heldInstance(now), OpHold, heldEntry{Instance: inst, Opened: now}); err != nil { return nil, false, fmt.Errorf("the record %s is opened but not listed: %w", inst, err) } + if read { + rs.sayFilling(inst, before) + } r, err = rs.Read(inst) if err == nil && r != nil { rs.keep(r) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go index 2a04a3214..87efb4b97 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go @@ -30,6 +30,15 @@ type ledgerFake struct { seq map[string]uint64 history []string failOn string + // fill is what Fill answers; nil, the ledger says no fill. + fill *buscalls.LedgerFilled +} + +func (f *ledgerFake) Fill() (buscalls.LedgerFilled, error) { + if f.fill == nil { + return buscalls.LedgerFilled{}, errors.New("the runtime answers no mesh/ledger.fill") + } + return *f.fill, nil } func newLedgerFake(t *testing.T) *ledgerFake { @@ -742,3 +751,119 @@ func TestTheRecordsToolsReadThemAndEndOneByHand(t *testing.T) { } }) } + +// W11.20 (design 47 *The delivery record*: a store filling is said before it is full): the gate reads its ledger's fill, +// says filling when a record takes it past 80 % of its cap, refuses the next record as full and holds its merge, and +// says room and hands the merge once the ledger is below 80 % again; the read tool says the fill. +func TestTheRecordsLedgerSaysFillingFullAndRoom(t *testing.T) { + const capBytes = 10 << 20 + filled := func(bytes uint64) *buscalls.LedgerFilled { + return &buscalls.LedgerFilled{History: buscalls.Fill{Count: 1, Bytes: bytes, MaxBytes: capBytes}, + State: buscalls.Fill{Bytes: 1, MaxBytes: 64 << 20}} + } + heads := map[string]string{"novox/app@111111111111": strings.Repeat("1", 40)} + at := time.Date(2026, 11, 9, 20, 0, 0, 0, time.UTC) + type said struct { + subject, id string + body meshdelivery.Fill + } + saying := func(out *[]said) SeatSay { + return func(subject string, body any, id string) error { + f, _ := body.(meshdelivery.Fill) + *out = append(*out, said{subject, id, f}) + return nil + } + } + + t.Run("filling once past 80 %, full refuses the next record, an unread fill refuses nothing", func(t *testing.T) { + l := newLedgerFake(t) + var events []said + rs := &Records{Ledger: l, Now: func() time.Time { return at }, Say: saying(&events)} + l.fill = filled(capBytes * 7 / 10) + r, opened, err := rs.Open("47/G", heads, false) + if err != nil || !opened { + t.Fatalf("open below the mark: %v %v", opened, err) + } + if len(events) != 0 { + t.Fatalf("said %+v below the mark", events) + } + // The record took the ledger past 80 %: said once as filling, with the store, its bytes and its cap. + l.fill = filled(capBytes * 85 / 100) + rs.sayFilling(r.Instance, buscalls.Fill{Bytes: capBytes * 7 / 10, MaxBytes: capBytes}) + if len(events) != 1 || events[0].subject != meshdelivery.SubjectFilling || events[0].body.Store != RecordsLedger || + events[0].body.Cap != capBytes || events[0].body.Bytes != capBytes*85/100 { + t.Fatalf("filling said as %+v", events) + } + // Already past the mark: a next record is opened and says nothing again. + events = nil + if _, opened, err := rs.Open("47/H", heads, false); err != nil || !opened || len(events) != 0 { + t.Fatalf("a record past the mark: %v %v, said %+v", opened, err, events) + } + // No room for one more record: refused, said as full naming what it refuses, nothing opened. + l.fill = filled(capBytes - RecordRoom/2) + n := len(l.history) + if _, opened, err := rs.Open("47/I", heads, false); !errors.Is(err, ErrRecordsFull) || opened || len(l.history) != n { + t.Fatalf("a full ledger opened %v (%v), %d operations", opened, err, len(l.history)-n) + } + if len(events) != 1 || events[0].subject != meshdelivery.SubjectFull || !strings.Contains(events[0].body.Refused, "47/I") { + t.Fatalf("full said as %+v", events) + } + if line, _ := rs.Lines(); !strings.HasPrefix(fmt.Sprint(line["fill"].(map[string]any)["state"]), "full") { + t.Fatalf("gate_records says the fill %v", line["fill"]) + } + // A ledger that says no fill (a runner before the call): the record is opened, and the read tool says why. + l.fill = nil + if _, opened, err := rs.Open("47/J", heads, false); err != nil || !opened { + t.Fatalf("an unread fill refused the record: %v %v", opened, err) + } + if line, _ := rs.Lines(); line["fill"].(map[string]any)["unread"] == nil { + t.Fatalf("gate_records says the unread fill as %v", line["fill"]) + } + }) + + t.Run("a merge refused for a full ledger is held, then handed with room once below 80 %", func(t *testing.T) { + w := approvedGroup(t) + l := newLedgerFake(t) + var events []said + w.h.Records = &Records{Ledger: l, Now: w.h.Now, Say: saying(&events)} + w.h.Merges = recordedMerges{MergeStore: w.merges, records: w.h.Records, members: w.h.groupMembers} + l.fill = filled(capBytes) + allGreen(w) + w.op.answer(w.op.open()[0], OptionApprove) + w.settleAll() + w.settleAll() + if len(w.forge.merges) != 0 { + t.Fatalf("a full ledger let %v merge", w.forge.merges) + } + w.h.mu.Lock() + since := w.h.heldFull() + w.h.mu.Unlock() + if since == nil || len(events) == 0 || events[0].subject != meshdelivery.SubjectFull { + t.Fatalf("the merge is not held for room (%v), said %+v", since, events) + } + // Room for a record but past 80 %: still held. + l.fill = filled(capBytes * 85 / 100) + w.later(askAgainAfter + time.Second) + w.settleAll() + if len(w.forge.merges) != 0 { + t.Fatalf("a ledger past 80 %% let %v merge after a full", w.forge.merges) + } + // Below 80 %: room said once, the held mark cleared, the group merged. + l.fill = filled(capBytes / 2) + w.later(askAgainAfter + time.Second) + w.settleAll() + w.settleAll() + rooms := 0 + for _, e := range events { + if e.subject == meshdelivery.SubjectRoom { + rooms++ + } + } + w.h.mu.Lock() + since = w.h.heldFull() + w.h.mu.Unlock() + if rooms != 1 || since != nil || len(w.forge.merges) == 0 { + t.Fatalf("room said %d times, held since %v, merges %v", rooms, since, w.forge.merges) + } + }) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordfill.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordfill.go new file mode 100644 index 000000000..33d856012 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordfill.go @@ -0,0 +1,201 @@ +package main + +// The delivery records' ledger against its cap (design 47 *The delivery record*: "A store filling is said before it is +// full"): the gate reads its ledger's fill through the tool runner (mesh-sdk buscalls.LedgerFill), says `filling` of the +// seat mesh-delivery when a record it opens takes the ledger past 80 % of its cap, refuses the next record when the +// ledger has no room for one and says `full`, and holds that record's merge until the ledger is below 80 % again, when +// it says `room` and hands the merge. Nothing is dropped to make room: only the retention removes a record. The fill +// is read from the bus at each judgement, and which merges wait for room is kept on their deliveries (Ready.HeldFull), +// so a restarted gate judges as the one before it. + +import ( + "encoding/json" + "errors" + "fmt" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// FillingPercent is the share of its cap past which the ledger is said filling, and below which room is said again. +const FillingPercent = 80 + +// RecordRoom is the room one record needs at most: an operation is at most one value of the bus. +const RecordRoom = 256 << 10 + +// ErrRecordsFull is a record refused because the records' ledger has no room for it. +var ErrRecordsFull = errors.New("the delivery records' ledger is full") + +// fillReader is a ledger that says its fill against its cap. +type fillReader interface { + Fill() (buscalls.LedgerFilled, error) +} + +func (sdkLedger) Fill() (buscalls.LedgerFilled, error) { + out, err := ledgerCall(buscalls.LedgerFill, buscalls.LedgerRequest{Ledger: RecordsLedger}) + if err != nil { + return buscalls.LedgerFilled{}, err + } + var got buscalls.LedgerFilled + if err := json.Unmarshal(out, &got); err != nil { + return got, fmt.Errorf("the ledger's fill is not readable: %w", err) + } + return got, nil +} + +// Fill is the records' ledger's store nearest its cap. +func (rs *Records) Fill() (buscalls.Fill, error) { + fr, ok := rs.Ledger.(fillReader) + if !ok { + return buscalls.Fill{}, errors.New("this gate's ledger says no fill") + } + f, err := fr.Fill() + if err != nil { + return buscalls.Fill{}, fmt.Errorf("the records' ledger cannot say its fill: %w", err) + } + return f.Fullest(), nil +} + +// pastFilling is a fill at or past FillingPercent of its cap. +func pastFilling(f buscalls.Fill) bool { + return f.MaxBytes > 0 && f.Bytes*100 >= f.MaxBytes*FillingPercent +} + +// noRoom is a fill that leaves no room for one more record. +func noRoom(f buscalls.Fill) bool { return f.MaxBytes > 0 && f.Bytes+RecordRoom > f.MaxBytes } + +// sayFill says filling, full or room of the seat under an id of the event and what it is about, so said again it is +// heard once. +func (rs *Records) sayFill(subject, about string, f buscalls.Fill, refused string) { + at := rs.now().UTC() + body := meshdelivery.Fill{Store: RecordsLedger, Bytes: int64(f.Bytes), Cap: int64(f.MaxBytes), Refused: refused, At: &at} + if err := rs.say(subject, body, about); err != nil { + rs.logf("[mesh-delivery] %s of the records' ledger (%d of %d bytes) is not said: %v", subject, f.Bytes, f.MaxBytes, err) + } +} + +func (rs *Records) logf(format string, args ...any) { + if rs.Logf != nil { + rs.Logf(format, args...) + } +} + +// refuseFull says full and answers ErrRecordsFull when the ledger has no room for the record inst; a fill it cannot +// read refuses nothing (the read tools say why it is not read). +func (rs *Records) refuseFull(milestone, inst string) (before buscalls.Fill, read bool, err error) { + before, fillErr := rs.Fill() + if fillErr != nil { + rs.logf("[mesh-delivery] the record %s is opened with its ledger's fill unread: %v", inst, fillErr) + return before, false, nil + } + if !noRoom(before) { + return before, true, nil + } + rs.sayFill(meshdelivery.SubjectFull, "full-"+inst, before, + "the next record, "+inst+" of "+milestone+"; its merge is held until room is said") + return before, true, fmt.Errorf("%w (%d of %d bytes): the record %s of %s is not opened, and its merge is held until "+ + "room is said: the setting record-retention or a larger cap is the operator's", ErrRecordsFull, before.Bytes, before.MaxBytes, inst, milestone) +} + +// sayFilling says filling when the record just opened took the ledger past FillingPercent of its cap. +func (rs *Records) sayFilling(inst string, before buscalls.Fill) { + if pastFilling(before) { + return + } + after, err := rs.Fill() + if err == nil && pastFilling(after) { + rs.sayFill(meshdelivery.SubjectFilling, "filling-"+inst, after, "") + } +} + +// RoomAgain is whether a merge held for a full ledger since then may be handed now: the records past their retention +// removed first, the ledger below FillingPercent of its cap, said once as room. A fill it cannot read gives no room. +func (rs *Records) RoomAgain(since time.Time) (bool, error) { + if err := rs.Sweep(rs.now()); err != nil { + return false, err + } + f, err := rs.Fill() + if err != nil { + return false, err + } + if pastFilling(f) { + return false, fmt.Errorf("the records' ledger holds %d of %d bytes, past %d %% of its cap: the merge waits for room", + f.Bytes, f.MaxBytes, FillingPercent) + } + rs.sayFill(meshdelivery.SubjectRoom, "room-"+since.UTC().Format(time.RFC3339Nano), f, "") + return true, nil +} + +// fillLine is the ledger's fill as the read tools say it. +func (rs *Records) fillLine() map[string]any { + f, err := rs.Fill() + if err != nil { + return map[string]any{"store": RecordsLedger, "unread": err.Error()} + } + line := map[string]any{"store": RecordsLedger, "bytes": f.Bytes, "cap": f.MaxBytes, "count": f.Count} + switch { + case noRoom(f): + line["state"] = "full: the next record is refused and its merge held" + case pastFilling(f): + line["state"] = fmt.Sprintf("filling: past %d %% of its cap", FillingPercent) + default: + line["state"] = "room" + } + return line +} + +// heldFull is the earliest time a merge was held for a full ledger, of every delivery that waits for room. Under the +// lock. +func (h *Holder) heldFull() *time.Time { + var since *time.Time + for _, d := range h.deliveries { + if r := d.Ready; r != nil && r.HeldFull != nil && (since == nil || r.HeldFull.Before(*since)) { + since = r.HeldFull + } + } + return since +} + +// roomFor clears every delivery's wait for room. Under the lock. +func (h *Holder) roomFor() { + for _, d := range h.deliveries { + if r := d.Ready; r != nil && r.HeldFull != nil { + r.HeldFull = nil + h.keep(d) + } + } +} + +// holdFull keeps that target's merge waits for room. Under the lock. +func (h *Holder) holdFull(target string, now time.Time) { + if d := h.deliveries[target]; d != nil && d.Ready != nil && d.Ready.HeldFull == nil { + at := now + d.Ready.HeldFull = &at + h.keep(d) + } +} + +// waitForRoom is whether a merge must wait because the records' ledger was full: while any merge waits for room, none +// is handed until RoomAgain gives it. +func (h *Holder) waitForRoom(target string) bool { + if h.Records == nil { + return false + } + h.mu.Lock() + since := h.heldFull() + h.mu.Unlock() + if since == nil { + return false + } + ok, err := h.Records.RoomAgain(*since) + if !ok { + h.giveBack(target, err) + return true + } + h.mu.Lock() + h.roomFor() + h.mu.Unlock() + h.Logf("[mesh-delivery] the records' ledger has room again: the merges held for it are handed") + return false +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordtools.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordtools.go index 72faf22d1..4941ed254 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordtools.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordtools.go @@ -105,7 +105,8 @@ func (rs *Records) Lines() (map[string]any, error) { } lines = append(lines, l) } - return map[string]any{"records": lines, "open": open, "kept": len(lines), "retention-days": int(rs.retention().Hours() / 24)}, nil + return map[string]any{"records": lines, "open": open, "kept": len(lines), "retention-days": int(rs.retention().Hours() / 24), + "fill": rs.fillLine()}, nil } // EndByHand ends one open record by hand, with the reason, recorded in the hand-act log first: nothing is ended when the @@ -161,7 +162,8 @@ func recordTools(h *Holder, acts handActs) []stdio.Tool { }}, {Name: "gate_records", Description: "Every delivery record kept, oldest first: its milestone, when it opened and ended, its entries, " + - "outcomes, nodes away and pins, its size against the bus's value limit, and while open what it waits for.", + "outcomes, nodes away and pins, its size against the bus's value limit, and while open what it waits for; and " + + "the records' ledger's fill against its cap (filling past 80 %, full when no record fits, else room).", Run: func(map[string]any) (any, error) { if h.Records == nil { return none() diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go index 8b4c14264..0a2ff4e52 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go @@ -38,22 +38,23 @@ type recordedMerges struct { members func(group string) []string } +// Put writes the group's merge into its delivery record first, so a ledger with no room refuses it before the merge is +// handed to the forge bridge (G §181), then into the state merges the bridge follows. func (m recordedMerges) Put(group string, v Merge) error { - if err := m.MergeStore.Put(group, v); err != nil { - return err + if m.records != nil && len(v.Approved) > 0 { + r, _, err := m.records.Open(group, v.Approved, false) + if err != nil { + return fmt.Errorf("the merge of %s is handed but not in its delivery record: %w", group, err) + } + var members []string + if m.members != nil { + members = m.members(group) + } + if err := m.records.SetGroup(r, recordGroupOf(v, members)); err != nil { + return err + } } - if m.records == nil || len(v.Approved) == 0 { - return nil - } - r, _, err := m.records.Open(group, v.Approved, false) - if err != nil { - return fmt.Errorf("the merge of %s is handed but not in its delivery record: %w", group, err) - } - var members []string - if m.members != nil { - members = m.members(group) - } - return m.records.SetGroup(r, recordGroupOf(v, members)) + return m.MergeStore.Put(group, v) } // switchGroup is one group in flight as the switch finds it: its milestone, its heads and its members.