diff --git a/go/stateobjects/send.go b/go/stateobjects/send.go index 6420a2ed..361588b1 100644 --- a/go/stateobjects/send.go +++ b/go/stateobjects/send.go @@ -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 } diff --git a/go/stateobjects/send_test.go b/go/stateobjects/send_test.go index fa8954c3..016f3601 100644 --- a/go/stateobjects/send_test.go +++ b/go/stateobjects/send_test.go @@ -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) + } +}