From c3c7bc90eb73e83f48fbeac6be149ade95a726c0 Mon Sep 17 00:00:00 2001 From: "Anderson A. Giacomolli" Date: Tue, 10 Nov 2026 02:06:24 -0300 Subject: [PATCH] Resume queued appends at their saved log ID Memory saves a queued message's position before it appends under the write lock. Recovery reads that position or completes its append, so a stop between the queue save and the log write cannot repeat the message. The queue owns the saved position and stops delivery after a failure. --- memory.go | 43 +++++++++++++++ memory_delivery_test.go | 117 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 160 insertions(+) create mode 100644 memory_delivery_test.go diff --git a/memory.go b/memory.go index c9fe1c5..a240b6c 100644 --- a/memory.go +++ b/memory.go @@ -149,6 +149,11 @@ func (m *Memory) AppendDated(kind, text, date string) (Message, error) { m.mutex.Lock() defer m.mutex.Unlock() + return m.appendDated(kind, text, date) +} + +// appendDated appends under the memory lock and updates the views and builds. +func (m *Memory) appendDated(kind, text, date string) (Message, error) { message, err := m.log.AppendDated(kind, text, date) if err != nil { return Message{}, fmt.Errorf("append dated log message: %w", err) @@ -172,6 +177,44 @@ func (m *Memory) AppendDated(kind, text, date string) (Message, error) { return message, nil } +// AppendOnce appends a queued message. prepare saves its log ID before the +// append, under the memory lock. A saved ID permits recovery after a stop +// between that save and the append, without a second log entry. prepare must +// not call Memory. The caller stops delivery after a failed append. +func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + if id != nil && *id < m.log.Len() { + message, err := m.log.Message(*id) + if err != nil { + return Message{}, fmt.Errorf("read queued message: %w", err) + } + + if message.Kind != kind || message.Text != text { + return Message{}, errors.New("queued message does not match its log entry") + } + + return message, nil + } + + if id != nil && *id != m.log.Len() { + return Message{}, errors.New("queued message is beyond the log") + } + + err := prepare(m.log.Len()) + if err != nil { + return Message{}, fmt.Errorf("prepare queued message: %w", err) + } + + message, err := m.appendDated(kind, text, "") + if err != nil { + return Message{}, fmt.Errorf("append queued message: %w", err) + } + + return message, nil +} + // children reads the two children of the node key. func (m *Memory) children(key NodeKey) (Node, Node, error) { var children [2]Node diff --git a/memory_delivery_test.go b/memory_delivery_test.go new file mode 100644 index 0000000..6f5e20c --- /dev/null +++ b/memory_delivery_test.go @@ -0,0 +1,117 @@ +package main + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestQueuedAppendResumesWithoutADuplicate(t *testing.T) { + dir := t.TempDir() + memory, err := openMemory(t.Context(), dir, newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + var saved *int + + first, err := memory.AppendOnce("user", "Record the gate size.", nil, func(id int) error { + saved = &id + + require.Equal(t, 0, memory.log.Len()) + + return nil + }) + require.NoError(t, err) + require.NoError(t, memory.Close()) + + memory, err = openMemory(t.Context(), dir, newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, memory.Close()) + }() + + second, err := memory.AppendOnce("user", first.Text, saved, func(int) error { + t.Fatal("a recorded message must not be prepared again") + + return nil + }) + require.NoError(t, err) + require.Equal(t, first, second) + require.Equal(t, 1, memory.Len()) +} + +func TestQueuedAppendRetriesAnUnwrittenPreparedMessage(t *testing.T) { + memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, memory.Close()) + }() + + failure := errors.New("fake queue write failure") + + var saved *int + + _, err = memory.AppendOnce("work", "[Garden] Check the gate.", nil, func(id int) error { + saved = &id + + return failure + }) + require.ErrorIs(t, err, failure) + require.Equal(t, 0, memory.Len()) + + message, err := memory.AppendOnce("work", "[Garden] Check the gate.", saved, func(id int) error { + require.Equal(t, *saved, id) + + return nil + }) + require.NoError(t, err) + require.Equal(t, 0, message.ID) + require.Equal(t, 1, memory.Len()) +} + +func TestQueuedAppendRefusesAnInvalidReceipt(t *testing.T) { + memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, memory.Close()) + }() + + _, err = memory.Append("user", "Keep the original.") + require.NoError(t, err) + + testCases := []struct { + name string + id int + cause string + }{ + { + name: "receipt points to different text", + id: 0, + cause: "does not match", + }, + { + name: "receipt skips a log position", + id: 2, + cause: "beyond the log", + }, + { + name: "receipt has a negative position", + id: -1, + cause: "read queued message", + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + _, err := memory.AppendOnce("user", "Other text.", &tc.id, func(int) error { + t.Fatal("an invalid receipt must not be prepared") + + return nil + }) + require.ErrorContains(t, err, tc.cause) + require.Equal(t, 1, memory.Len()) + }) + } +} -- 2.53.0