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 +}