Handle reuploads (#1049)

Update the mirror to permit a small number of bundles to be re-uploaded by a client, as permitted by the tlog-mirror spec, and updates the POSIX mirror storage to write bundles in an idempotent fashion.

Author: AlCutter

mirror_lifecycle.go

@@ -54,6 +54,11 @@ var (
 	ErrInvalidProof = errors.New("invalid proof")
 )
 
+// maxExcessEntries is the maximum number of "excess" entries that can be
+// re-uploaded. This is intended to prevent mirror re-uploads from going too far
+// back in time.
+const maxExcessEntries = 2048
+
 // MirrorOptions holds mirror lifecycle settings for all storage implementations.
 type MirrorOptions struct {
 	signer      note.Signer
@@ -113,6 +118,7 @@ func (o *MirrorOptions) valid() error {
 type MirrorWriter interface {
 	// IntegrateBundles integrates bundles of log entries, starting at the given bundle index, into the local tree.
 	// Bundles are _always_ aligned on bundle boundaries.
+	// Implementations MUST NOT overwrite entries that are already integrated into the tree.
 	//
 	// Returns the size of the tree and its new root hash if successful.
 	// If the provided iterator yields an error, the MirrorWriter MUST return it either directly, or wrapped so the caller can identify it.
@@ -224,11 +230,10 @@ func (mt *MirrorTarget) AddEntries(ctx context.Context, uploadStart, uploadEnd u
 	//    - upload_start:
 	//      * MUST NOT be greater than the mirror's next expected entry index.
 	//      * MUST NOT be too far below the mirror's next entry index.
-	if (uploadStart == 0 && uploadEnd == 0 && !userTicketValid) ||
+	if excessEntries := min(uploadEnd, nextEntry) - uploadStart; (uploadStart == 0 && uploadEnd == 0 && !userTicketValid) ||
 		(uploadEnd != pendingSize || uploadEnd < nextEntry) ||
-		(uploadStart > nextEntry) {
-		// TODO(al): add flexibility about re-writing some entries
-		slog.ErrorContext(ctx, "Returning conflict", slog.Uint64("nextEntry", nextEntry), slog.Uint64("pendingSize", pendingSize), slog.Uint64("uploadStart", uploadStart), slog.Uint64("uploadEnd", uploadEnd))
+		(uploadStart > nextEntry || excessEntries > maxExcessEntries) {
+		slog.ErrorContext(ctx, "Returning conflict", slog.Bool("ticket_valid", userTicketValid), slog.Uint64("next_entry", nextEntry), slog.Uint64("pending_size", pendingSize), slog.Uint64("upload_start", uploadStart), slog.Uint64("upload_end", uploadEnd), slog.Uint64("excess_entries", excessEntries))
 		return nextEntry, pendingSize, ticketBytes, nil, ErrConflict
 	}
 
@@ -269,6 +274,7 @@ func (mt *MirrorTarget) bundleIterator(ctx context.Context, next func() (*Mirror
 	crf := compact.RangeFactory{Hash: rfc6962.DefaultHasher.HashChildren}
 	return func(yield func(*api.EntryBundle, error) bool) {
 		// Check for unaligned upload start, and fetch entries from the start of the bundle to use to pad.
+		// This is necessary for the subtree proof for such an unaligned first bundle to validate.
 		var padEntries [][]byte
 		if p := start % layout.EntryBundleWidth; p != 0 {
 			// non-aligned starting bundle

mirror_lifecycle_test.go

@@ -629,3 +629,82 @@ func TestMirrorTarget_AddEntries_NoPendingCheckpoint(t *testing.T) {
 		t.Errorf("got error %v, want ErrNoPendingCheckpoint", err)
 	}
 }
+
+func TestMirrorTarget_AddEntries_UploadStartConflicts(t *testing.T) {
+	const (
+		testIntegratedSize = uint64(3072)
+		testUploadEnd      = uint64(4096)
+	)
+
+	ctx := t.Context()
+	testPendingCPCustom := mustSignCP(testPendingCPOrigin, testUploadEnd, testPendingCPRoot, testLogSigner)
+
+	for _, tc := range []struct {
+		name         string
+		uploadStart  uint64
+		wantConflict bool
+	}{
+		{
+			name:         "exact start",
+			uploadStart:  testIntegratedSize,
+			wantConflict: false,
+		},
+		{
+			name:         "re-upload within limit",
+			uploadStart:  testIntegratedSize - maxExcessEntries,
+			wantConflict: false,
+		},
+		{
+			name:         "re-upload too far back",
+			uploadStart:  testIntegratedSize - maxExcessEntries - 1,
+			wantConflict: true,
+		},
+		{
+			name:         "uploadStart > nextEntry",
+			uploadStart:  testIntegratedSize + 256,
+			wantConflict: true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			mt := &MirrorTarget{
+				logVerifier: testLogVerifier,
+				signer:      testMirrorSigner,
+				writer: &fakeMirrorWriter{
+					integrateFunc: func(ctx context.Context, fromBundleIdx uint64, bundles iter.Seq2[*api.EntryBundle, error]) (uint64, []byte, error) {
+						for _, err := range bundles {
+							if err != nil {
+								return 0, nil, err
+							}
+						}
+						pendingCPRoot, err := base64.StdEncoding.DecodeString(testPendingCPRoot)
+						return testUploadEnd, pendingCPRoot, err
+					},
+					updateCheckpointFunc: func(ctx context.Context, f func(oldCP []byte) (newCP []byte, err error)) error {
+						return nil
+					},
+				},
+				reader: &fakeLogReader{
+					sizeFunc: func(ctx context.Context) (uint64, error) { return testIntegratedSize, nil },
+				},
+				cpSource: func(ctx context.Context) ([]byte, error) { return []byte(testPendingCPCustom), nil },
+			}
+
+			validTicket, err := mt.seal([]byte(testPendingCPCustom))
+			if err != nil {
+				t.Fatalf("seal failed: %v", err)
+			}
+
+			_, _, _, _, err = mt.AddEntries(ctx, tc.uploadStart, testUploadEnd, validTicket, func() (*MirrorPackage, error) {
+				return nil, io.EOF
+			})
+
+			if gotConflict := errors.Is(err, ErrConflict); gotConflict != tc.wantConflict {
+				t.Fatalf("AddEntries got conflict %v, want conflict %v (err: %v)", gotConflict, tc.wantConflict, err)
+			}
+			if !tc.wantConflict && err != nil {
+				t.Fatalf("AddEntries unexpected error: %v", err)
+			}
+
+		})
+	}
+}

storage/posix/file_ops.go

@@ -15,6 +15,7 @@
 package posix
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -127,13 +128,34 @@ func createEx(name string, d []byte) error {
 		}()
 
 		if err := os.Link(tmpName, name); err != nil {
-			// Wrap the error here because we need to know if it's os.ErrExists at higher levels.
+			// Wrap the error here because we need to know if it's os.ErrExist at higher levels.
 			return fmt.Errorf("failed to link temporary file to target %q: %w", name, err)
 		}
 		return nil
 	})
 }
 
+// createIdempotent atomically creates a file at the given path containing the provided data, and syncs the
+// directory containing the newly created file.
+//
+// Returns an error if a file already exists at the specified location and it contents are not identical to
+// the provided data, or if it's unable to fully write the data & close the file.
+func createIdempotent(name string, d []byte) error {
+	if err := createEx(name, d); err != nil {
+		if !errors.Is(err, os.ErrExist) {
+			return err
+		}
+		e, err := os.ReadFile(name)
+		if err != nil {
+			return fmt.Errorf("failed to read existing file: %w", err)
+		}
+		if !bytes.Equal(e, d) {
+			return fmt.Errorf("existing file %q has different contents", name)
+		}
+	}
+	return nil
+}
+
 // overwrite atomically creates/overwrites a file at the given path containing the provided data, and syncs
 // the directory containing the overwritten/created file.
 func overwrite(name string, d []byte) error {

storage/posix/files.go

@@ -528,6 +528,17 @@ func (lrs *logResourceStorage) writeBundle(ctx context.Context, index uint64, pa
 	})
 }
 
+// writeBundleIdempotent takes care of writing out the serialised entry bundle file if it doesn't already exist.
+func (lrs *logResourceStorage) writeBundleIdempotent(ctx context.Context, index uint64, partial uint8, bundle []byte) error {
+	return otel.TraceErr(ctx, "tessera.storage.posix.writeBundleIdempotent", tracer, func(ctx context.Context, span trace.Span) error {
+		bf := lrs.entriesPath(index, partial)
+		if err := lrs.s.createIdempotent(bf, bundle); err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
 // initialise ensures that the storage location is valid by loading the checkpoint from this location, or
 // creating a zero-sized one if it doesn't already exist.
 func (a *appender) initialise(ctx context.Context) error {
@@ -915,6 +926,12 @@ func (s *Storage) createOverwrite(p string, d []byte) error {
 	return overwrite(filepath.Join(s.cfg.Path, p), d)
 }
 
+// createIdempotent atomically creates a file at the given path with the provided data, if it doesn't already exist.
+// Returns an error if the target path already exists but contains different data.
+func (s *Storage) createIdempotent(p string, d []byte) error {
+	return createIdempotent(filepath.Join(s.cfg.Path, p), d)
+}
+
 func (s *Storage) readAll(p string) ([]byte, error) {
 	p = filepath.Join(s.cfg.Path, p)
 	return os.ReadFile(p)
@@ -1236,7 +1253,7 @@ func (m *MirrorWriter) IntegrateBundles(ctx context.Context, bundleIdx uint64, b
 		}
 
 		// Now write the bundle out.
-		if err := m.logStorage.writeBundle(ctx, bundleIdx, uint8(p), bundleData); err != nil {
+		if err := m.logStorage.writeBundleIdempotent(ctx, bundleIdx, uint8(p), bundleData); err != nil {
 			return 0, nil, err
 		}
 

storage/posix/files_test.go

@@ -676,3 +676,84 @@ func TestOverwrite_CleanupOnRenameFailure(t *testing.T) {
 		}
 	}
 }
+
+func TestMirrorWriter_IntegrateBundles(t *testing.T) {
+	baseBundle := &api.EntryBundle{
+		Entries: [][]byte{[]byte("entry1"), []byte("entry2")},
+	}
+
+	bundlesFunc := func(b *api.EntryBundle) func(yield func(*api.EntryBundle, error) bool) {
+		return func(yield func(*api.EntryBundle, error) bool) {
+			yield(b, nil)
+		}
+	}
+
+	for _, tc := range []struct {
+		name         string
+		bundleIdx    uint64
+		bundle       *api.EntryBundle
+		expectedSize uint64
+		wantErr      bool
+		errSubstring string
+	}{
+		{
+			name:         "integrate bundle with common prefix (extend)",
+			bundleIdx:    0,
+			bundle:       &api.EntryBundle{Entries: append(append([][]byte{}, baseBundle.Entries...), []byte("entry3"))},
+			expectedSize: 3,
+		},
+		{
+			name:         "re-integrate base bundle (idempotency)",
+			bundleIdx:    0,
+			bundle:       baseBundle,
+			expectedSize: 2,
+		},
+		{
+			name:         "re-integrate modified basebundle (conflict error)",
+			bundleIdx:    0,
+			bundle:       &api.EntryBundle{Entries: append(append([][]byte{}, baseBundle.Entries[:len(baseBundle.Entries)-1]...), []byte("entry2_modified"))},
+			wantErr:      true,
+			errSubstring: "different contents",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s := &Storage{
+				cfg: Config{
+					HTTPClient: http.DefaultClient,
+					Path:       t.TempDir(),
+				},
+			}
+			opts := tessera.NewMirrorOptions()
+			ctx := t.Context()
+			mw, _, err := s.MirrorWriter(ctx, opts)
+			if err != nil {
+				t.Fatalf("MirrorWriter: %v", err)
+			}
+
+			// Set up the storage with the base bundle.
+			if size, _, err := mw.IntegrateBundles(ctx, 0, bundlesFunc(baseBundle)); err != nil {
+				t.Fatalf("Failed to set up test, IntegrateBundles: %v", err)
+			} else if size != uint64(len(baseBundle.Entries)) {
+				t.Fatalf("Failed to set up test, expected size %d, got %d", len(baseBundle.Entries), size)
+			}
+
+			// Now run the test case.
+			size, _, err := mw.IntegrateBundles(ctx, tc.bundleIdx, bundlesFunc(tc.bundle))
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("expected error containing %q, got nil", tc.errSubstring)
+				}
+				if !strings.Contains(err.Error(), tc.errSubstring) {
+					t.Fatalf("expected error containing %q, got %v", tc.errSubstring, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if size != tc.expectedSize {
+				t.Fatalf("got size %d, want %d", size, tc.expectedSize)
+			}
+		})
+	}
+}