From 20876b85e13b01d7049b51e6f04a686fbb123431 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Thu, 10 Sep 2026 18:05:43 +0200 Subject: [PATCH] refactor: stream store records as events Stream now yields a store.Event sum type instead of a RevocationRecord, so revocation and principal invalidation records can share one poll, cursor, settle window and dedup map. Event carries a Kind and one of Revocation or PrincipalRevocation, with Cause and RecordedAt accessors for the fields the stream loop needs. Nothing produces principal events yet. The memory backend keeps its map for Get and streams from an append-only log, so every stored record is streamed in the order it was stored. The postgres backend wraps the rows it already reads. The firehose handler switches on the event kind and refuses kinds it has no wire format for. The memory backend logs one entry per cause link, so a retried revocation invocation streams once, as the postgres backend's ON CONFLICT (id) DO NOTHING already gives. Co-Authored-By: Claude Fable 5.1 --- pkg/fx/app.go | 15 ++++- pkg/fx/app_test.go | 6 +- pkg/store/memory/store.go | 105 +++++++++++++++++++++---------- pkg/store/memory/store_test.go | 83 +++++++++++++++++++++--- pkg/store/postgres/store.go | 35 ++++++----- pkg/store/postgres/store_test.go | 53 +++++++++++----- pkg/store/store.go | 76 +++++++++++++++++++++- pkg/store/store_test.go | 58 +++++++++++++++++ 8 files changed, 347 insertions(+), 84 deletions(-) create mode 100644 pkg/store/store_test.go diff --git a/pkg/fx/app.go b/pkg/fx/app.go index 9ed7279..a619e15 100644 --- a/pkg/fx/app.go +++ b/pkg/fx/app.go @@ -312,7 +312,7 @@ 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 @@ -320,7 +320,7 @@ func firehoseHandler(revocations store.RevocationStore) echo.HandlerFunc { 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 { diff --git a/pkg/fx/app_test.go b/pkg/fx/app_test.go index 7057fcb..509c558 100644 --- a/pkg/fx/app_test.go +++ b/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) } } diff --git a/pkg/store/memory/store.go b/pkg/store/memory/store.go index ce52c5f..a7af8c7 100644 --- a/pkg/store/memory/store.go +++ b/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 +} diff --git a/pkg/store/memory/store_test.go b/pkg/store/memory/store_test.go index bc7d8f6..762d884 100644 --- a/pkg/store/memory/store_test.go +++ b/pkg/store/memory/store_test.go @@ -57,12 +57,12 @@ 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) @@ -70,13 +70,38 @@ func TestMemoryRevocationStoreStream(t *testing.T) { // 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 { diff --git a/pkg/store/postgres/store.go b/pkg/store/postgres/store.go index d788efc..fb78688 100644 --- a/pkg/store/postgres/store.go +++ b/pkg/store/postgres/store.go @@ -102,9 +102,9 @@ func (s *Store) Get(ctx context.Context, revoked cid.Cid) (store.RevocationRecor return 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) { ticker := time.NewTicker(streamPollInterval) defer ticker.Stop() @@ -119,29 +119,32 @@ func (s *Store) Stream(ctx context.Context, from time.Time) iter.Seq2[store.Revo seen := map[cid.Cid]time.Time{} for { if err := ctx.Err(); err != nil { - yield(store.RevocationRecord{}, err) + yield(store.Event{}, err) return } // The horizon is read before the rows so it never overtakes them. var dbNow time.Time if err := s.pool.QueryRow(ctx, `SELECT now()`).Scan(&dbNow); err != nil { - yield(store.RevocationRecord{}, fmt.Errorf("reading database clock: %w", err)) + yield(store.Event{}, fmt.Errorf("reading database clock: %w", err)) return } for rec, err := range s.recordsFrom(ctx, cursor) { if err != nil { - yield(store.RevocationRecord{}, err) + yield(store.Event{}, err) return } - link := rec.Cause.Link() + link := rec.Cause().Link() if _, ok := seen[link]; ok { continue } + // Read before yielding: the consumer may retain and mutate + // the event it is handed. + recordedAt := rec.RecordedAt() if !yield(rec, nil) { return } - seen[link] = rec.RecordedAt + seen[link] = recordedAt } if horizon := dbNow.Add(-streamSettleWindow); horizon.After(cursor) { cursor = horizon @@ -154,7 +157,7 @@ func (s *Store) Stream(ctx context.Context, from time.Time) iter.Seq2[store.Revo select { case <-ctx.Done(): - yield(store.RevocationRecord{}, ctx.Err()) + yield(store.Event{}, ctx.Err()) return case <-ticker.C: } @@ -162,8 +165,8 @@ func (s *Store) Stream(ctx context.Context, from time.Time) iter.Seq2[store.Revo } } -func (s *Store) recordsFrom(ctx context.Context, cursor time.Time) iter.Seq2[store.RevocationRecord, error] { - return func(yield func(store.RevocationRecord, error) bool) { +func (s *Store) recordsFrom(ctx context.Context, cursor time.Time) iter.Seq2[store.Event, error] { + return func(yield func(store.Event, error) bool) { // The query is inclusive at the cursor: each poll re-reads the // unsettled window and Stream skips rows it already yielded. The // zero cursor predates every record, so an unbounded stream matches @@ -177,7 +180,7 @@ func (s *Store) recordsFrom(ctx context.Context, cursor time.Time) iter.Seq2[sto cursor, ) if err != nil { - yield(store.RevocationRecord{}, fmt.Errorf("querying revocations: %w", err)) + yield(store.Event{}, fmt.Errorf("querying revocations: %w", err)) return } defer rows.Close() @@ -188,20 +191,20 @@ func (s *Store) recordsFrom(ctx context.Context, cursor time.Time) iter.Seq2[sto var pathWitness [][]byte var recordedAt time.Time if err := rows.Scan(&causeBytes, &revoke, &pathWitness, &recordedAt); err != nil { - yield(store.RevocationRecord{}, fmt.Errorf("scanning revocation: %w", err)) + yield(store.Event{}, fmt.Errorf("scanning revocation: %w", err)) return } record, err := decodeRecord(causeBytes, revoke, pathWitness, recordedAt) if err != nil { - yield(store.RevocationRecord{}, fmt.Errorf("decoding revocation: %w", err)) + yield(store.Event{}, fmt.Errorf("decoding revocation: %w", err)) return } - if !yield(record, nil) { + if !yield(store.RevocationEvent(record), nil) { return } } if err := rows.Err(); err != nil { - yield(store.RevocationRecord{}, fmt.Errorf("iterating revocations: %w", err)) + yield(store.Event{}, fmt.Errorf("iterating revocations: %w", err)) } } } diff --git a/pkg/store/postgres/store_test.go b/pkg/store/postgres/store_test.go index 05cfee4..0d1a8b6 100644 --- a/pkg/store/postgres/store_test.go +++ b/pkg/store/postgres/store_test.go @@ -60,12 +60,12 @@ func TestPostgresRevocationStoreStream(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) @@ -73,13 +73,32 @@ func TestPostgresRevocationStoreStream(t *testing.T) { // 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 TestPostgresRevocationStoreStreamEvents(t *testing.T) { + s, _ := newTestStore(t) + 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()) + cancel() + require.ErrorIs(t, <-done, context.Canceled) +} + func TestPostgresRevocationStoreStreamSameTimestamp(t *testing.T) { s, pool := newTestStore(t) recordedAt := time.Now().UTC().Truncate(time.Microsecond) @@ -95,8 +114,8 @@ func TestPostgresRevocationStoreStreamSameTimestamp(t *testing.T) { // Records sharing recorded_at are ordered by id, so delivery order is not // deterministic: both must arrive, exactly once each. delivered := map[string]int{} - delivered[(<-records).Cause.Link().String()]++ - delivered[(<-records).Cause.Link().String()]++ + delivered[(<-records).Cause().Link().String()]++ + delivered[(<-records).Cause().Link().String()]++ require.Equal(t, 1, delivered[firstRevocation.Link().String()]) require.Equal(t, 1, delivered[secondRevocation.Link().String()]) @@ -104,7 +123,7 @@ func TestPostgresRevocationStoreStreamSameTimestamp(t *testing.T) { // timestamp must not be re-delivered. select { case record := <-records: - require.Failf(t, "stream re-delivered a record", "cause: %s", record.Cause.Link()) + require.Failf(t, "stream re-delivered a record", "cause: %s", record.Cause().Link()) case <-time.After(2 * time.Second): } cancel() @@ -120,7 +139,7 @@ func TestPostgresRevocationStoreStreamLateArrivals(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, firstRevocation.Link(), (<-records).Cause().Link()) // A concurrent insert can commit a row at a timestamp the stream has // already passed: sharing the first record's recorded_at, or earlier. @@ -131,15 +150,15 @@ func TestPostgresRevocationStoreStreamLateArrivals(t *testing.T) { insertAt(t, pool, earlierRevocation, earlierPath, recordedAt.Add(-time.Second)) delivered := map[string]int{} - delivered[(<-records).Cause.Link().String()]++ - delivered[(<-records).Cause.Link().String()]++ + delivered[(<-records).Cause().Link().String()]++ + delivered[(<-records).Cause().Link().String()]++ require.Equal(t, 1, delivered[sameTsRevocation.Link().String()]) require.Equal(t, 1, delivered[earlierRevocation.Link().String()]) // Hold the stream open past a poll interval: nothing is re-delivered. select { case record := <-records: - require.Failf(t, "stream re-delivered a record", "cause: %s", record.Cause.Link()) + require.Failf(t, "stream re-delivered a record", "cause: %s", record.Cause().Link()) case <-time.After(2 * time.Second): } cancel() @@ -160,8 +179,8 @@ func TestPostgresRevocationStoreStreamBroadcasts(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() @@ -250,8 +269,8 @@ func insertAt(t *testing.T, pool *pgxpool.Pool, revocation ucan.Invocation, path require.NoError(t, err) } -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 { diff --git a/pkg/store/store.go b/pkg/store/store.go index 2139797..5d8d702 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -5,6 +5,7 @@ import ( "iter" "time" + "github.com/fil-forge/ucantone/did" "github.com/fil-forge/ucantone/errors" "github.com/fil-forge/ucantone/ucan" "github.com/ipfs/go-cid" @@ -27,6 +28,75 @@ type RevocationRecord struct { RecordedAt time.Time } +// PrincipalRevocationRecord records that every proof cached for a principal's +// keys is void. It names the principal and no delegation. +type PrincipalRevocationRecord struct { + // Tenant is the DID of the tenant the principal belongs to. + Tenant did.DID + // Principal is the principal's identifier, unique within the tenant. + Principal string + // Cause is the invocation that invalidated the principal. + Cause ucan.Invocation + // RecordedAt is the time when the record was recorded. Note this is not + // necessarily the time when the invalidation was issued. + RecordedAt time.Time +} + +// EventKind names the kind of record an [Event] carries. +type EventKind string + +const ( + // EventKindRevocation marks an event carrying a [RevocationRecord]. + EventKindRevocation EventKind = "revocation" + // EventKindPrincipalRevocation marks an event carrying a + // [PrincipalRevocationRecord]. + EventKindPrincipalRevocation EventKind = "principal" +) + +// Event is one record streamed from the store. Exactly one of Revocation and +// PrincipalRevocation is set, as named by Kind. +type Event struct { + Kind EventKind + Revocation *RevocationRecord + PrincipalRevocation *PrincipalRevocationRecord +} + +// RevocationEvent wraps a revocation record as an [Event]. +func RevocationEvent(record RevocationRecord) Event { + return Event{Kind: EventKindRevocation, Revocation: &record} +} + +// PrincipalRevocationEvent wraps a principal revocation record as an [Event]. +func PrincipalRevocationEvent(record PrincipalRevocationRecord) Event { + return Event{Kind: EventKindPrincipalRevocation, PrincipalRevocation: &record} +} + +// Cause returns the invocation that caused the event's record, or nil when +// the event carries no record. +func (e Event) Cause() ucan.Invocation { + switch { + case e.Revocation != nil: + return e.Revocation.Cause + case e.PrincipalRevocation != nil: + return e.PrincipalRevocation.Cause + default: + return nil + } +} + +// RecordedAt returns the time the event's record was recorded, or the zero +// time when the event carries no record. +func (e Event) RecordedAt() time.Time { + switch { + case e.Revocation != nil: + return e.Revocation.RecordedAt + case e.PrincipalRevocation != nil: + return e.PrincipalRevocation.RecordedAt + default: + return time.Time{} + } +} + type RevocationStore interface { // Add adds a revocation record to the store. The path is the delegation chain // from the root delegation to the revoked delegation. The issuer of the @@ -35,10 +105,10 @@ type RevocationStore interface { // Get retrieves a revocation record from the store by revoked delegation CID. // If the record is not found, [ErrNotFound] is returned. Get(ctx context.Context, revoked cid.Cid) (RevocationRecord, error) - // Stream streams all revocation records from the store and remains open until - // the context is canceled. The from parameter filters records to those + // Stream streams all records from the store as events and remains open + // until the context is canceled. The from parameter filters records to those // recorded on or after the given time, so consumers resuming from the // timestamp of the last record they received do not miss records that share // it. If from is zero, all records are returned. - Stream(ctx context.Context, from time.Time) iter.Seq2[RevocationRecord, error] + Stream(ctx context.Context, from time.Time) iter.Seq2[Event, error] } diff --git a/pkg/store/store_test.go b/pkg/store/store_test.go new file mode 100644 index 0000000..76bc671 --- /dev/null +++ b/pkg/store/store_test.go @@ -0,0 +1,58 @@ +package store_test + +import ( + "testing" + "time" + + "github.com/fil-forge/swarf/pkg/store" + "github.com/fil-forge/ucantone/did" + "github.com/fil-forge/ucantone/multikey/ed25519" + "github.com/fil-forge/ucantone/ucan/command" + "github.com/fil-forge/ucantone/ucan/invocation" + "github.com/stretchr/testify/require" +) + +func TestRevocationEvent(t *testing.T) { + cause := invoke(t) + recordedAt := time.Now() + event := store.RevocationEvent(store.RevocationRecord{Cause: cause, RecordedAt: recordedAt}) + + require.Equal(t, store.EventKindRevocation, event.Kind) + require.NotNil(t, event.Revocation) + require.Nil(t, event.PrincipalRevocation) + require.Equal(t, cause.Link(), event.Cause().Link()) + require.True(t, event.RecordedAt().Equal(recordedAt)) +} + +func TestPrincipalRevocationEvent(t *testing.T) { + cause := invoke(t) + recordedAt := time.Now() + tenant, err := did.Parse("did:plc:tenant") + require.NoError(t, err) + event := store.PrincipalRevocationEvent(store.PrincipalRevocationRecord{Tenant: tenant, Principal: "alice", Cause: cause, RecordedAt: recordedAt}) + + require.Equal(t, store.EventKindPrincipalRevocation, event.Kind) + require.Nil(t, event.Revocation) + require.NotNil(t, event.PrincipalRevocation) + require.Equal(t, tenant, event.PrincipalRevocation.Tenant) + require.Equal(t, "alice", event.PrincipalRevocation.Principal) + require.Equal(t, cause.Link(), event.Cause().Link()) + require.True(t, event.RecordedAt().Equal(recordedAt)) +} + +func TestEmptyEvent(t *testing.T) { + var event store.Event + require.Nil(t, event.Cause()) + require.True(t, event.RecordedAt().IsZero()) +} + +func invoke(t *testing.T) *invocation.Invocation { + t.Helper() + issuer, err := ed25519.GenerateIssuer() + require.NoError(t, err) + cmd, err := command.Parse("/test/invoke") + require.NoError(t, err) + inv, err := invocation.Invoke(issuer, did.Undef, cmd, nil) + require.NoError(t, err) + return inv +}