refactor: stream store records as events

pkg/fx/app.go

@@ -312,15 +312,15 @@ func firehoseHandler(revocations store.RevocationStore) echo.HandlerFunc {
 		response.Header().Set(echo.HeaderConnection, "keep-alive")
 		response.WriteHeader(http.StatusOK)
 
-		for record, err := range revocations.Stream(c.Request().Context(), from) {
+		for event, err := range revocations.Stream(c.Request().Context(), from) {
 			if err != nil {
 				if errors.Is(err, context.Canceled) {
 					return nil
 				}
 				writeFirehoseError(response, err)
 				return nil
 			}
-			if err := writeFirehoseRecord(response, record); err != nil {
+			if err := writeFirehoseEvent(response, event); err != nil {
 				return err
 			}
 		}
@@ -361,6 +361,17 @@ func parseFrom(value string) (time.Time, error) {
 	return from, nil
 }
 
+// writeFirehoseEvent writes one server-sent event for a streamed store event.
+// Event kinds without a wire representation are an error.
+func writeFirehoseEvent(response *echo.Response, event store.Event) error {
+	switch event.Kind {
+	case store.EventKindRevocation:
+		return writeFirehoseRecord(response, *event.Revocation)
+	default:
+		return fmt.Errorf("unsupported firehose event kind %q", event.Kind)
+	}
+}
+
 func writeFirehoseRecord(response *echo.Response, record store.RevocationRecord) error {
 	path := make([]cid.Cid, len(record.Path))
 	for i, delegation := range record.Path {

pkg/fx/app_test.go

@@ -230,9 +230,9 @@ func (s *firehoseTestStore) Get(context.Context, cid.Cid) (store.RevocationRecor
 	return s.record, nil
 }
 
-func (s *firehoseTestStore) Stream(_ context.Context, from time.Time) iter.Seq2[store.RevocationRecord, error] {
+func (s *firehoseTestStore) Stream(_ context.Context, from time.Time) iter.Seq2[store.Event, error] {
 	s.from = from
-	return func(yield func(store.RevocationRecord, error) bool) {
-		yield(s.record, nil)
+	return func(yield func(store.Event, error) bool) {
+		yield(store.RevocationEvent(s.record), nil)
 	}
 }

pkg/store/memory/store.go

@@ -16,22 +16,30 @@ import (
 
 // Store stores revocation records in process memory.
 type Store struct {
-	mu             sync.RWMutex
-	records        map[cid.Cid]memoryRecord
+	mu sync.RWMutex
+	// records indexes the latest revocation record by revoked delegation for Get.
+	records map[cid.Cid]store.RevocationRecord
+	// log is the append-only sequence of every stored record, in the order it
+	// was stored, for Stream. Entry i has seq i+1.
+	log []logEntry
+	// causes holds the cause link of every logged record, so a retried
+	// invocation is logged once.
+	causes         map[cid.Cid]struct{}
 	subscribers    map[uint64]chan struct{}
 	nextSeq        uint64
 	nextSubscriber uint64
 }
 
-type memoryRecord struct {
-	record store.RevocationRecord
-	seq    uint64
+type logEntry struct {
+	event store.Event
+	seq   uint64
 }
 
 // New creates an empty revocation store.
 func New() *Store {
 	return &Store{
-		records:     make(map[cid.Cid]memoryRecord),
+		records:     make(map[cid.Cid]store.RevocationRecord),
+		causes:      make(map[cid.Cid]struct{}),
 		subscribers: make(map[uint64]chan struct{}),
 	}
 }
@@ -48,30 +56,48 @@ func (s *Store) Add(ctx context.Context, revocation ucan.Invocation, path []ucan
 	}
 
 	record := store.RevocationRecord{
-		Revoke:     path[len(path)-1].Link(),
-		Cause:      revocation,
-		Path:       append([]ucan.Delegation(nil), path...),
-		RecordedAt: time.Now(),
+		Revoke: path[len(path)-1].Link(),
+		Cause:  revocation,
+		Path:   append([]ucan.Delegation(nil), path...),
 	}
 
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	// Stamped under the lock so the log's append order is its time order:
+	// a streamer that has already yielded a later record never sees an
+	// earlier one arrive behind it.
+	record.RecordedAt = time.Now()
+
 	if s.records == nil {
-		s.records = make(map[cid.Cid]memoryRecord)
+		s.records = make(map[cid.Cid]store.RevocationRecord)
 	}
-	s.nextSeq++
-	s.records[path[len(path)-1].Link()] = memoryRecord{
-		record: record,
-		seq:    s.nextSeq,
+	s.records[record.Revoke] = record
+	s.appendLocked(store.RevocationEvent(record))
+	return nil
+}
+
+// appendLocked appends event to the log and wakes every streamer. An event
+// whose cause is already logged is dropped, so a retried invocation is
+// streamed once, as the postgres backend's ON CONFLICT (id) DO NOTHING gives.
+// The caller must hold the write lock.
+func (s *Store) appendLocked(event store.Event) {
+	cause := event.Cause().Link()
+	if _, ok := s.causes[cause]; ok {
+		return
+	}
+	if s.causes == nil {
+		s.causes = make(map[cid.Cid]struct{})
 	}
+	s.causes[cause] = struct{}{}
+	s.nextSeq++
+	s.log = append(s.log, logEntry{event: event, seq: s.nextSeq})
 	for _, notification := range s.subscribers {
 		select {
 		case notification <- struct{}{}:
 		default:
 		}
 	}
-	return nil
 }
 
 // Get retrieves the revocation record for delegation.
@@ -81,25 +107,25 @@ func (s *Store) Get(ctx context.Context, delegation cid.Cid) (store.RevocationRe
 	}
 
 	s.mu.RLock()
-	entry, ok := s.records[delegation]
+	record, ok := s.records[delegation]
 	s.mu.RUnlock()
 	if !ok {
 		return store.RevocationRecord{}, store.ErrNotFound
 	}
 
-	return copyRecord(entry.record), nil
+	return copyRecord(record), nil
 }
 
-// Stream returns matching revocation records and remains open until ctx is canceled.
-func (s *Store) Stream(ctx context.Context, from time.Time) iter.Seq2[store.RevocationRecord, error] {
-	return func(yield func(store.RevocationRecord, error) bool) {
+// Stream returns matching records as events and remains open until ctx is canceled.
+func (s *Store) Stream(ctx context.Context, from time.Time) iter.Seq2[store.Event, error] {
+	return func(yield func(store.Event, error) bool) {
 		if err := ctx.Err(); err != nil {
-			yield(store.RevocationRecord{}, err)
+			yield(store.Event{}, err)
 			return
 		}
 
 		s.mu.Lock()
-		entries, sequence := s.recordsFromLocked(0, from)
+		entries, sequence := s.eventsFromLocked(0, from)
 		s.nextSubscriber++
 		subscriber := s.nextSubscriber
 		notification := make(chan struct{}, 1)
@@ -117,39 +143,42 @@ func (s *Store) Stream(ctx context.Context, from time.Time) iter.Seq2[store.Revo
 		for {
 			for _, entry := range entries {
 				if err := ctx.Err(); err != nil {
-					yield(store.RevocationRecord{}, err)
+					yield(store.Event{}, err)
 					return
 				}
-				if !yield(copyRecord(entry.record), nil) {
+				if !yield(copyEvent(entry.event), nil) {
 					return
 				}
 			}
 
 			select {
 			case <-ctx.Done():
-				yield(store.RevocationRecord{}, ctx.Err())
+				yield(store.Event{}, ctx.Err())
 				return
 			case <-notification:
 				s.mu.RLock()
-				entries, sequence = s.recordsFromLocked(sequence, from)
+				entries, sequence = s.eventsFromLocked(sequence, from)
 				s.mu.RUnlock()
 			}
 		}
 	}
 }
 
-func (s *Store) recordsFromLocked(sequence uint64, from time.Time) ([]memoryRecord, uint64) {
-	entries := make([]memoryRecord, 0, len(s.records))
-	for _, entry := range s.records {
-		if entry.seq > sequence && (from.IsZero() || !entry.record.RecordedAt.Before(from)) {
+// eventsFromLocked returns the log entries appended after sequence and
+// recorded on or after from, ordered by recorded time then sequence, along
+// with the sequence to resume from. The caller must hold the lock.
+func (s *Store) eventsFromLocked(sequence uint64, from time.Time) ([]logEntry, uint64) {
+	entries := make([]logEntry, 0, len(s.log)-int(sequence))
+	for _, entry := range s.log[sequence:] {
+		if from.IsZero() || !entry.event.RecordedAt().Before(from) {
 			entries = append(entries, entry)
 		}
 	}
 	sort.Slice(entries, func(i, j int) bool {
-		if entries[i].record.RecordedAt.Equal(entries[j].record.RecordedAt) {
+		if entries[i].event.RecordedAt().Equal(entries[j].event.RecordedAt()) {
 			return entries[i].seq < entries[j].seq
 		}
-		return entries[i].record.RecordedAt.Before(entries[j].record.RecordedAt)
+		return entries[i].event.RecordedAt().Before(entries[j].event.RecordedAt())
 	})
 	return entries, s.nextSeq
 }
@@ -158,3 +187,13 @@ func copyRecord(record store.RevocationRecord) store.RevocationRecord {
 	record.Path = slices.Clone(record.Path)
 	return record
 }
+
+func copyEvent(event store.Event) store.Event {
+	if event.Revocation != nil {
+		return store.RevocationEvent(copyRecord(*event.Revocation))
+	}
+	if event.PrincipalRevocation != nil {
+		return store.PrincipalRevocationEvent(*event.PrincipalRevocation)
+	}
+	return event
+}

pkg/store/memory/store_test.go

@@ -57,26 +57,51 @@ func TestMemoryRevocationStoreStream(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	records, done := collectStream(s.Stream(ctx, time.Time{}))
-	require.Equal(t, firstRevocation.Link(), (<-records).Cause.Link())
-	require.Equal(t, secondRevocation.Link(), (<-records).Cause.Link())
+	require.Equal(t, firstRevocation.Link(), (<-records).Cause().Link())
+	require.Equal(t, secondRevocation.Link(), (<-records).Cause().Link())
 
 	thirdRevocation, thirdPath := revocationPath(t)
 	add(t, s, thirdRevocation, thirdPath)
-	require.Equal(t, thirdRevocation.Link(), (<-records).Cause.Link())
+	require.Equal(t, thirdRevocation.Link(), (<-records).Cause().Link())
 	cancel()
 	require.ErrorIs(t, <-done, context.Canceled)
 
 	// The from filter is inclusive: the record recorded at exactly from is
 	// re-delivered.
 	filteredCtx, filteredCancel := context.WithCancel(context.Background())
 	filtered, filteredDone := collectStream(s.Stream(filteredCtx, first.RecordedAt))
-	require.Equal(t, firstRevocation.Link(), (<-filtered).Cause.Link())
-	require.Equal(t, secondRevocation.Link(), (<-filtered).Cause.Link())
-	require.Equal(t, thirdRevocation.Link(), (<-filtered).Cause.Link())
+	require.Equal(t, firstRevocation.Link(), (<-filtered).Cause().Link())
+	require.Equal(t, secondRevocation.Link(), (<-filtered).Cause().Link())
+	require.Equal(t, thirdRevocation.Link(), (<-filtered).Cause().Link())
 	filteredCancel()
 	require.ErrorIs(t, <-filteredDone, context.Canceled)
 }
 
+func TestMemoryRevocationStoreStreamEvents(t *testing.T) {
+	s := memory.New()
+	revocation, path := revocationPath(t)
+	add(t, s, revocation, path)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	events, done := collectStream(s.Stream(ctx, time.Time{}))
+	event := <-events
+	require.Equal(t, store.EventKindRevocation, event.Kind)
+	require.Nil(t, event.PrincipalRevocation)
+	require.NotNil(t, event.Revocation)
+	require.Equal(t, path[len(path)-1].Link(), event.Revocation.Revoke)
+	require.Equal(t, revocation.Link(), event.Revocation.Cause.Link())
+	require.Len(t, event.Revocation.Path, len(path))
+	require.False(t, event.Revocation.RecordedAt.IsZero())
+
+	// Streamed records are copies: mutating one does not affect the store.
+	event.Revocation.Path[0] = nil
+	again, err := s.Get(context.Background(), path[len(path)-1].Link())
+	require.NoError(t, err)
+	require.NotNil(t, again.Path[0])
+	cancel()
+	require.ErrorIs(t, <-done, context.Canceled)
+}
+
 func TestMemoryRevocationStoreStreamBroadcasts(t *testing.T) {
 	s := memory.New()
 	firstRevocation, firstPath := revocationPath(t)
@@ -91,8 +116,8 @@ func TestMemoryRevocationStoreStreamBroadcasts(t *testing.T) {
 
 	revocation, path := revocationPath(t)
 	add(t, s, revocation, path)
-	require.Equal(t, revocation.Link(), (<-firstRecords).Cause.Link())
-	require.Equal(t, revocation.Link(), (<-secondRecords).Cause.Link())
+	require.Equal(t, revocation.Link(), (<-firstRecords).Cause().Link())
+	require.Equal(t, revocation.Link(), (<-secondRecords).Cause().Link())
 
 	firstCancel()
 	secondCancel()
@@ -114,6 +139,44 @@ func TestMemoryRevocationStoreStreamCanceled(t *testing.T) {
 	require.Fail(t, "Stream did not return context.Canceled")
 }
 
+func TestMemoryRevocationStoreAddIgnoresRetriedRevocation(t *testing.T) {
+	s := memory.New()
+	revocation, path := revocationPath(t)
+	add(t, s, revocation, path)
+	add(t, s, revocation, path)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	events, done := collectStream(s.Stream(ctx, time.Time{}))
+	require.Equal(t, revocation.Link(), (<-events).Cause().Link())
+	select {
+	case event := <-events:
+		require.Fail(t, "retried revocation streamed twice", event.Cause().Link())
+	case <-time.After(100 * time.Millisecond):
+	}
+	cancel()
+	require.ErrorIs(t, <-done, context.Canceled)
+}
+
+func TestMemoryRevocationStoreAddSecondRevocationOfDelegation(t *testing.T) {
+	s := memory.New()
+	firstRevocation, path := revocationPath(t)
+	// A second revocation of the same delegation, by a different invocation.
+	secondRevocation, _ := revocationPath(t)
+	add(t, s, firstRevocation, path)
+	add(t, s, secondRevocation, path)
+
+	record, err := s.Get(context.Background(), path[len(path)-1].Link())
+	require.NoError(t, err)
+	require.Equal(t, secondRevocation.Link(), record.Cause.Link())
+
+	ctx, cancel := context.WithCancel(context.Background())
+	events, done := collectStream(s.Stream(ctx, time.Time{}))
+	require.Equal(t, firstRevocation.Link(), (<-events).Cause().Link())
+	require.Equal(t, secondRevocation.Link(), (<-events).Cause().Link())
+	cancel()
+	require.ErrorIs(t, <-done, context.Canceled)
+}
+
 func TestMemoryRevocationStoreAddRejectsInvalidPath(t *testing.T) {
 	err := memory.New().Add(context.Background(), nil, nil)
 	require.Error(t, err)
@@ -124,8 +187,8 @@ func add(t *testing.T, s store.RevocationStore, revocation ucan.Invocation, path
 	require.NoError(t, s.Add(context.Background(), revocation, path))
 }
 
-func collectStream(stream iter.Seq2[store.RevocationRecord, error]) (<-chan store.RevocationRecord, <-chan error) {
-	records := make(chan store.RevocationRecord)
+func collectStream(stream iter.Seq2[store.Event, error]) (<-chan store.Event, <-chan error) {
+	records := make(chan store.Event)
 	done := make(chan error, 1)
 	go func() {
 		for record, err := range stream {