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.

diff --git a/memory.go b/memory.go
index c9fe1c567504c11f0a94be48fdb581298cde510c..a240b6c16d1d898e3051f3e0a0c7a19434061b3e 100644 (file)
--- 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()
 
        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)
        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
 }
 
        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
 // 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 (file)
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())
+               })
+       }
+}