Memoize the outer-most AddFn (#711)

Author: AlCutter

append_lifecycle.go

@@ -258,11 +258,28 @@ func NewAppender(ctx context.Context, d Driver, opts *AppendOptions) (*Appender,
 		ctx, span := tracer.Start(ctx, "tessera.Appender.Add")
 		defer span.End()
 
-		return t.Add(ctx, entry)
+		// NOTE: We memoize the returned value here so that repeated calls to the returned
+		//		 future don't result in unexpected side-effects from inner AddFn functions
+		//		 being called multiple times.
+		//		 Currently this is the outermost wrapping of Add so we do the memoization
+		//		 here, if this changes, ensure that we move the memoization call so that
+		//		 this remains true.
+		return memoizeFuture(t.Add(ctx, entry))
 	}
 	return a, t.Shutdown, r, nil
 }
 
+// memoizeFuture wraps an AddFn delegate with logic to ensure that the delegate is called at most
+// once.
+func memoizeFuture(delegate IndexFuture) IndexFuture {
+	f := sync.OnceValues(func() (Index, error) {
+		return delegate()
+	})
+	return func() (Index, error) {
+		return f()
+	}
+}
+
 func followerStats(ctx context.Context, f Follower, size func(context.Context) (uint64, error)) {
 	name := f.Name()
 	t := time.NewTicker(200 * time.Millisecond)

append_lifecycle_test.go

@@ -0,0 +1,56 @@
+// Copyright 2025 The Tessera authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tessera
+
+import (
+	"context"
+	"testing"
+)
+
+func TestMemoize(t *testing.T) {
+	// Set up an AddFn which will increment a counter every time it's called, and return that in the Index.
+	i := uint64(0)
+	deleg := func() (Index, error) {
+		i++
+		return Index{
+			Index: i,
+		}, nil
+	}
+	add := func(_ context.Context, _ *Entry) IndexFuture {
+		return deleg
+	}
+
+	// Create a single future (for a single Entry), and convince ourselves that the counter is being incremented
+	// each time the future is being invoked.
+	f1 := add(nil, nil)
+	a, _ := f1()
+	b, _ := f1()
+	if a.Index == b.Index {
+		t.Fatalf("a(=%d) == b(=%d)", a.Index, b.Index)
+	}
+
+	// Now create an AddFn which memoizes the result of the delegate, like we do in NewAppender, and assert that
+	// repeated calls to the future work as expected; only incrementing the counter once.
+	add = func(_ context.Context, _ *Entry) IndexFuture {
+		return memoizeFuture(deleg)
+	}
+	f2 := add(nil, nil)
+	c, _ := f2()
+	d, _ := f2()
+
+	if c.Index != d.Index {
+		t.Fatalf("c(=%d) != d(=%d)", c.Index, d.Index)
+	}
+}

client/stream_test.go

@@ -42,10 +42,10 @@ func TestEntryBundles(t *testing.T) {
 		}
 	}()
 
-	if _, err := populateEntries(ctx, tl, logSize1, "first"); err != nil {
+	if _, err := populateEntries(t, tl, logSize1, "first"); err != nil {
 		t.Fatalf("populateEntries(first): %v", err)
 	}
-	if _, err := populateEntries(ctx, tl, logSize2-logSize1, "second"); err != nil {
+	if _, err := populateEntries(t, tl, logSize2-logSize1, "second"); err != nil {
 		t.Fatalf("populateEntries(second): %v", err)
 	}
 
@@ -95,7 +95,7 @@ func TestEntries(t *testing.T) {
 	}()
 
 	// Put some entries into a log.
-	es, err := populateEntries(ctx, tl, logSize, "first")
+	es, err := populateEntries(t, tl, logSize, "first")
 	if err != nil {
 		t.Fatalf("populateEntries(): %v", err)
 	}
@@ -140,18 +140,21 @@ func TestEntries(t *testing.T) {
 	}
 }
 
-func populateEntries(ctx context.Context, tl *testonly.TestLog, N uint64, ep string) ([][]byte, error) {
+func populateEntries(t *testing.T, tl *testonly.TestLog, N uint64, ep string) ([][]byte, error) {
+	t.Helper()
+
 	es := make([][]byte, 0, N)
 	fs := make([]tessera.IndexFuture, 0, N)
 	for i := range N {
 		e := fmt.Appendf(nil, "%s-%d", ep, i)
 		es = append(es, e)
-		fs = append(fs, tl.Appender.Add(ctx, tessera.NewEntry(e)))
+		fs = append(fs, tl.Appender.Add(t.Context(), tessera.NewEntry(e)))
 	}
+	t.Logf("Added %d entries", N)
 
-	a := tessera.NewPublicationAwaiter(ctx, tl.LogReader.ReadCheckpoint, time.Second)
+	a := tessera.NewPublicationAwaiter(t.Context(), tl.LogReader.ReadCheckpoint, time.Second)
 	for _, f := range fs {
-		if _, _, err := a.Await(ctx, f); err != nil {
+		if _, _, err := a.Await(t.Context(), f); err != nil {
 			return nil, err
 		}
 	}