O-8: a foundation part's sender keeps its send sequence and ages its sent store from it
Builds: design 47 G §330, §410, §59 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -2,7 +2,9 @@ package stateobjects
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
||||
@@ -25,6 +27,69 @@ type Sender struct {
|
||||
Store string
|
||||
// Limit is the connection's largest message; zero is MessageLimit.
|
||||
Limit int
|
||||
// Log is a foundation part's old `state` section `sent-log`, with Node the key of this node's sends in it and Now
|
||||
// the clock (zero is time.Now): the sender keeps the sequence of its own sends there and ages the store from it at
|
||||
// each send (design 47 G §410, §330). A sender with no Log ages nothing: the bus expires a template store's objects.
|
||||
Log KV
|
||||
Node string
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// The sent store of a foundation part keeps an object SentMaxAge, and a send deletes at most SentAgeAtMost of the
|
||||
// objects past it (design 47 G §330, §59).
|
||||
const (
|
||||
SentLogState = "sent-log"
|
||||
SentAgeAtMost = 20
|
||||
)
|
||||
|
||||
type sentEntry struct {
|
||||
Object string `json:"object"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// record writes this send into the sequence of the sender's sends by compare-and-set, and deletes the objects of the
|
||||
// oldest sends that are past SentMaxAge, at most SentAgeAtMost; two sends at once lose none, the one that lost the
|
||||
// compare reading the sequence again.
|
||||
func (s Sender) record(object string) error {
|
||||
now := time.Now
|
||||
if s.Now != nil {
|
||||
now = s.Now
|
||||
}
|
||||
for {
|
||||
e, err := s.Log.Get(s.Node)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var log []sentEntry
|
||||
if e != nil {
|
||||
if err := json.Unmarshal(e.Value, &log); err != nil {
|
||||
return fmt.Errorf("the %s of %s: %w", SentLogState, s.Node, err)
|
||||
}
|
||||
}
|
||||
var aged []sentEntry
|
||||
for len(aged) < SentAgeAtMost && len(log) > 0 && now().Sub(log[0].At) > SentMaxAge {
|
||||
aged, log = append(aged, log[0]), log[1:]
|
||||
}
|
||||
log = append(log, sentEntry{object, now()})
|
||||
raw, _ := json.Marshal(log)
|
||||
if e == nil {
|
||||
err = s.Log.Create(s.Node, raw)
|
||||
} else {
|
||||
err = s.Log.Update(s.Node, raw, e.Revision)
|
||||
}
|
||||
if errors.Is(err, ErrChanged) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, a := range aged {
|
||||
if err := s.Objects.Delete(a.Object); err != nil {
|
||||
return fmt.Errorf("aging the sent object %s: %w", a.Object, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// Send is what to send for body: the body itself when it fits, else the reference of the object it was put in as
|
||||
@@ -47,6 +112,11 @@ func (s Sender) Send(label string, body []byte) ([]byte, error) {
|
||||
if _, err := putObject(s.Objects, ref.Object, body); err != nil {
|
||||
return nil, fmt.Errorf("the object for %s: %w", label, err)
|
||||
}
|
||||
if s.Log != nil {
|
||||
if err := s.record(ref.Object); err != nil {
|
||||
return nil, fmt.Errorf("the %s for %s: %w", SentLogState, label, err)
|
||||
}
|
||||
}
|
||||
return Wrap(ref), nil
|
||||
}
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ADR 0666 §16: a reply or an event over the per-message limit carries the name of an object, and the receiver reads it
|
||||
@@ -66,3 +67,31 @@ func TestABundleReadsASentObjectThroughTheRuntime(t *testing.T) {
|
||||
t.Fatalf("the bundle answered %v", out)
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §410, §330: a foundation part's sender keeps the sequence of its sends in its old state section
|
||||
// `sent-log` and deletes the sent objects past 7 days at a later send, at most 20 a send.
|
||||
func TestAFoundationPartsSentStoreAgesFromItsSendLog(t *testing.T) {
|
||||
m := newMemory()
|
||||
clock := time.Date(2026, 11, 1, 0, 0, 0, 0, time.UTC)
|
||||
s := Sender{Objects: memObjects{m}, Store: "m_sent-objects", Log: m, Node: "n1", Now: func() time.Time { return clock }}
|
||||
var first string
|
||||
for i := 0; i < 25; i++ {
|
||||
sent, err := s.Send("reply."+string(rune('a'+i)), bytes.Repeat([]byte{byte('a' + i)}, 2<<20))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ref, _ := ReadRef(sent); i == 0 {
|
||||
first = ref.Object
|
||||
}
|
||||
}
|
||||
if len(m.objects) != 25 {
|
||||
t.Fatalf("%d objects within 7 days, want 25", len(m.objects))
|
||||
}
|
||||
clock = clock.Add(SentMaxAge + time.Hour)
|
||||
if _, err := s.Send("reply.late", bytes.Repeat([]byte("z"), 2<<20)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, kept := m.objects[first]; kept || len(m.objects) != 6 {
|
||||
t.Fatalf("after a late send %d objects remain (the first kept: %v), want 6", len(m.objects), kept)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user