Check missed events before start (#5289)

Signed-off-by: Faisal Memon <fymemon@yahoo.com>

pkg/server/datastore/datastore.go

@@ -172,6 +172,7 @@ type ListAttestedNodesResponse struct {
 
 type ListAttestedNodesEventsRequest struct {
  GreaterThanEventID uint
+ LessThanEventID uint
 }
 
 type AttestedNodeEvent struct {
@@ -180,8 +181,7 @@ type AttestedNodeEvent struct {
 }
 
 type ListAttestedNodesEventsResponse struct {
- Events []AttestedNodeEvent
- FirstEventID uint
+ Events []AttestedNodeEvent
 }
 
 type ListBundlesRequest struct {
@@ -226,6 +226,7 @@ type ListRegistrationEntriesResponse struct {
 
 type ListRegistrationEntriesEventsRequest struct {
  GreaterThanEventID uint
+ LessThanEventID uint
 }
 
 type RegistrationEntryEvent struct {
@@ -234,8 +235,7 @@ type RegistrationEntryEvent struct {
 }
 
 type ListRegistrationEntriesEventsResponse struct {
- Events []RegistrationEntryEvent
- FirstEventID uint
+ Events []RegistrationEntryEvent
 }
 
 type ListFederationRelationshipsRequest struct {

pkg/server/datastore/sqlstore/sqlstore.go

@@ -1719,8 +1719,20 @@ func createAttestedNodeEvent(tx *gorm.DB, event *datastore.AttestedNodeEvent) er
 
 func listAttestedNodesEvents(tx *gorm.DB, req *datastore.ListAttestedNodesEventsRequest) (*datastore.ListAttestedNodesEventsResponse, error) {
  var events []AttestedNodeEvent
- if err := tx.Find(&events, "id > ?", req.GreaterThanEventID).Order("id asc").Error; err != nil {
- return nil, sqlError.Wrap(err)
+
+ if req.GreaterThanEventID != 0 || req.LessThanEventID != 0 {
+ query, id, err := buildListEventsQueryString(req.GreaterThanEventID, req.LessThanEventID)
+ if err != nil {
+ return nil, sqlError.Wrap(err)
+ }
+
+ if err := tx.Find(&events, query.String(), id).Order("id asc").Error; err != nil {
+ return nil, sqlError.Wrap(err)
+ }
+ } else {
+ if err := tx.Find(&events).Order("id asc").Error; err != nil {
+ return nil, sqlError.Wrap(err)
+ }
  }
 
  resp := &datastore.ListAttestedNodesEventsResponse{
@@ -1730,9 +1742,6 @@ func listAttestedNodesEvents(tx *gorm.DB, req *datastore.ListAttestedNodesEvents
  resp.Events[i].EventID = event.ID
  resp.Events[i].SpiffeID = event.SpiffeID
  }
- if len(events) > 0 {
- resp.FirstEventID = events[0].ID
- }
 
  return resp, nil
 }
@@ -4096,8 +4105,20 @@ func deleteRegistrationEntryEvent(tx *gorm.DB, eventID uint) error {
 
 func listRegistrationEntriesEvents(tx *gorm.DB, req *datastore.ListRegistrationEntriesEventsRequest) (*datastore.ListRegistrationEntriesEventsResponse, error) {
  var events []RegisteredEntryEvent
- if err := tx.Find(&events, "id > ?", req.GreaterThanEventID).Order("id asc").Error; err != nil {
- return nil, sqlError.Wrap(err)
+
+ if req.GreaterThanEventID != 0 || req.LessThanEventID != 0 {
+ query, id, err := buildListEventsQueryString(req.GreaterThanEventID, req.LessThanEventID)
+ if err != nil {
+ return nil, sqlError.Wrap(err)
+ }
+
+ if err := tx.Find(&events, query.String(), id).Order("id asc").Error; err != nil {
+ return nil, sqlError.Wrap(err)
+ }
+ } else {
+ if err := tx.Find(&events).Order("id asc").Error; err != nil {
+ return nil, sqlError.Wrap(err)
+ }
  }
 
  resp := &datastore.ListRegistrationEntriesEventsResponse{
@@ -4107,9 +4128,6 @@ func listRegistrationEntriesEvents(tx *gorm.DB, req *datastore.ListRegistrationE
  resp.Events[i].EventID = event.ID
  resp.Events[i].EntryID = event.EntryID
  }
- if len(events) > 0 {
- resp.FirstEventID = events[0].ID
- }
 
  return resp, nil
 }
@@ -4122,6 +4140,26 @@ func pruneRegistrationEntriesEvents(tx *gorm.DB, olderThan time.Duration) error
  return nil
 }
 
+func buildListEventsQueryString(greaterThanEventID, lessThanEventID uint) (*strings.Builder, uint, error) {
+ if greaterThanEventID != 0 && lessThanEventID != 0 {
+ return nil, 0, errors.New("can't set both greater and less than event id")
+ }
+
+ var id uint
+ query := new(strings.Builder)
+ query.WriteString("id ")
+ if greaterThanEventID != 0 {
+ query.WriteString("> ?")
+ id = greaterThanEventID
+ }
+ if lessThanEventID != 0 {
+ query.WriteString("< ?")
+ id = lessThanEventID
+ }
+
+ return query, id, nil
+}
+
 func createJoinToken(tx *gorm.DB, token *datastore.JoinToken) error {
  t := JoinToken{
  Token: token.Token,

pkg/server/datastore/sqlstore/sqlstore_test.go

@@ -1539,36 +1539,62 @@ func (s *PluginSuite) TestListAttestedNodesEvents() {
  tests := []struct {
  name string
  greaterThanEventID uint
+ lessThanEventID uint
  expectedEvents []datastore.AttestedNodeEvent
  expectedFirstEventID uint
+ expectedLastEventID uint
+ expectedErr string
  }{
  {
  name: "All Events",
  greaterThanEventID: 0,
  expectedFirstEventID: 1,
+ expectedLastEventID: uint(len(expectedEvents)),
  expectedEvents: expectedEvents,
  },
  {
- name: "Half of the Events",
+ name: "Greater than half of the Events",
  greaterThanEventID: uint(len(expectedEvents) / 2),
  expectedFirstEventID: uint(len(expectedEvents)/2) + 1,
+ expectedLastEventID: uint(len(expectedEvents)),
  expectedEvents: expectedEvents[len(expectedEvents)/2:],
  },
  {
- name: "None of the Events",
- greaterThanEventID: uint(len(expectedEvents)),
- expectedFirstEventID: 0,
- expectedEvents: []datastore.AttestedNodeEvent{},
+ name: "Less than half of the Events",
+ lessThanEventID: uint(len(expectedEvents) / 2),
+ expectedFirstEventID: 1,
+ expectedLastEventID: uint(len(expectedEvents)/2) - 1,
+ expectedEvents: expectedEvents[:len(expectedEvents)/2-1],
+ },
+ {
+ name: "Greater than largest Event ID",
+ greaterThanEventID: uint(len(expectedEvents)),
+ expectedEvents: []datastore.AttestedNodeEvent{},
+ },
+ {
+ name: "Setting both greater and less than",
+ greaterThanEventID: 1,
+ lessThanEventID: 1,
+ expectedErr: "rpc error: code = Unknown desc = datastore-sql: can't set both greater and less than event id",
  },
  }
  for _, test := range tests {
  s.T().Run(test.name, func(t *testing.T) {
  resp, err := s.ds.ListAttestedNodesEvents(ctx, &datastore.ListAttestedNodesEventsRequest{
  GreaterThanEventID: test.greaterThanEventID,
+ LessThanEventID: test.lessThanEventID,
  })
- s.Assert().NoError(err)
- s.Assert().Equal(test.expectedFirstEventID, resp.FirstEventID)
- s.Assert().Equal(test.expectedEvents, resp.Events)
+ if test.expectedErr != "" {
+ require.EqualError(t, err, test.expectedErr)
+ return
+ }
+ s.Require().NoError(err)
+
+ s.Require().Equal(test.expectedEvents, resp.Events)
+ if len(resp.Events) > 0 {
+ s.Require().Equal(test.expectedFirstEventID, resp.Events[0].EventID)
+ s.Require().Equal(test.expectedLastEventID, resp.Events[len(resp.Events)-1].EventID)
+ }
  })
  }
 }
@@ -3998,36 +4024,62 @@ func (s *PluginSuite) TestListRegistrationEntriesEvents() {
  tests := []struct {
  name string
  greaterThanEventID uint
+ lessThanEventID uint
  expectedEvents []datastore.RegistrationEntryEvent
  expectedFirstEventID uint
+ expectedLastEventID uint
+ expectedErr string
  }{
  {
  name: "All Events",
  greaterThanEventID: 0,
  expectedFirstEventID: 1,
+ expectedLastEventID: uint(len(expectedEvents)),
  expectedEvents: expectedEvents,
  },
  {
- name: "Half of the Events",
- greaterThanEventID: 2,
- expectedFirstEventID: 3,
- expectedEvents: expectedEvents[2:],
+ name: "Greater than half of the Events",
+ greaterThanEventID: uint(len(expectedEvents) / 2),
+ expectedFirstEventID: uint(len(expectedEvents)/2) + 1,
+ expectedLastEventID: uint(len(expectedEvents)),
+ expectedEvents: expectedEvents[len(expectedEvents)/2:],
+ },
+ {
+ name: "Less than half of the Events",
+ lessThanEventID: uint(len(expectedEvents) / 2),
+ expectedFirstEventID: 1,
+ expectedLastEventID: uint(len(expectedEvents)/2) - 1,
+ expectedEvents: expectedEvents[:len(expectedEvents)/2-1],
+ },
+ {
+ name: "Greater than largest Event ID",
+ greaterThanEventID: 4,
+ expectedEvents: []datastore.RegistrationEntryEvent{},
  },
  {
- name:  "None of the Events",
- greaterThanEventID:  4,
- expectedFirstEventID: 0,
- expectedEvents: []datastore.RegistrationEntryEvent{},
+ name: "Setting both greater and less than",
+ greaterThanEventID: 1,
+ lessThanEventID: 1,
+ expectedErr:  "rpc error: code = Unknown desc = datastore-sql: can't set both greater and less than event id",
  },
  }
  for _, test := range tests {
  s.T().Run(test.name, func(t *testing.T) {
  resp, err = s.ds.ListRegistrationEntriesEvents(ctx, &datastore.ListRegistrationEntriesEventsRequest{
  GreaterThanEventID: test.greaterThanEventID,
+ LessThanEventID: test.lessThanEventID,
  })
+ if test.expectedErr != "" {
+ require.EqualError(t, err, test.expectedErr)
+ return
+ }
  s.Require().NoError(err)
- s.Require().Equal(test.expectedFirstEventID, resp.FirstEventID)
+
  s.Require().Equal(test.expectedEvents, resp.Events)
+ if len(resp.Events) > 0 {
+ s.Require().Equal(test.expectedFirstEventID, resp.Events[0].EventID)
+ s.Require().Equal(test.expectedLastEventID, resp.Events[len(resp.Events)-1].EventID)
+ }
  })
  }
 }

pkg/server/endpoints/authorized_entryfetcher.go

@@ -24,37 +24,35 @@ type AuthorizedEntryFetcherWithEventsBasedCache struct {
  log logrus.FieldLogger
  ds datastore.DataStore
 
- cacheReloadInterval time.Duration
- pruneEventsOlderThan time.Duration
- sqlTransactionTimeout time.Duration
+ cacheReloadInterval time.Duration
+ pruneEventsOlderThan time.Duration
 
  registrationEntries eventsBasedCache
  attestedNodes eventsBasedCache
 }
 
 type eventsBasedCache interface {
  updateCache(ctx context.Context) error
- pruneMissedEvents(sqlTransactionTimeout time.Duration)
+ pruneMissedEvents()
 }
 
 func NewAuthorizedEntryFetcherWithEventsBasedCache(ctx context.Context, log logrus.FieldLogger, clk clock.Clock, ds datastore.DataStore, cacheReloadInterval, pruneEventsOlderThan, sqlTransactionTimeout time.Duration) (*AuthorizedEntryFetcherWithEventsBasedCache, error) {
  log.Info("Building event-based in-memory entry cache")
- cache, registrationEntries, attestedNodes, err := buildCache(ctx, log, ds, clk)
+ cache, registrationEntries, attestedNodes, err := buildCache(ctx, log, ds, clk, sqlTransactionTimeout)
  if err != nil {
  return nil, err
  }
  log.Info("Completed building event-based in-memory entry cache")
 
  return &AuthorizedEntryFetcherWithEventsBasedCache{
- cache: cache,
- clk: clk,
- log: log,
- ds: ds,
- cacheReloadInterval: cacheReloadInterval,
- pruneEventsOlderThan: pruneEventsOlderThan,
- sqlTransactionTimeout: sqlTransactionTimeout,
- registrationEntries: registrationEntries,
- attestedNodes: attestedNodes,
+ cache: cache,
+ clk: clk,
+ log: log,
+ ds: ds,
+ cacheReloadInterval: cacheReloadInterval,
+ pruneEventsOlderThan: pruneEventsOlderThan,
+ registrationEntries: registrationEntries,
+ attestedNodes: attestedNodes,
  }, nil
 }
 
@@ -100,8 +98,8 @@ func (a *AuthorizedEntryFetcherWithEventsBasedCache) pruneEvents(ctx context.Con
  pruneRegistrationEntriesEventsErr := a.ds.PruneRegistrationEntriesEvents(ctx, olderThan)
  pruneAttestedNodesEventsErr := a.ds.PruneAttestedNodesEvents(ctx, olderThan)
 
- a.registrationEntries.pruneMissedEvents(a.sqlTransactionTimeout)
- a.attestedNodes.pruneMissedEvents(a.sqlTransactionTimeout)
+ a.registrationEntries.pruneMissedEvents()
+ a.attestedNodes.pruneMissedEvents()
 
  return errors.Join(pruneRegistrationEntriesEventsErr, pruneAttestedNodesEventsErr)
 }
@@ -113,15 +111,15 @@ func (a *AuthorizedEntryFetcherWithEventsBasedCache) updateCache(ctx context.Con
  return errors.Join(updateRegistrationEntriesCacheErr, updateAttestedNodesCacheErr)
 }
 
-func buildCache(ctx context.Context, log logrus.FieldLogger, ds datastore.DataStore, clk clock.Clock) (*authorizedentries.Cache, *registrationEntries, *attestedNodes, error) {
+func buildCache(ctx context.Context, log logrus.FieldLogger, ds datastore.DataStore, clk clock.Clock, sqlTransactionTimeout time.Duration) (*authorizedentries.Cache, *registrationEntries, *attestedNodes, error) {
  cache := authorizedentries.NewCache(clk)
 
- registrationEntries, err := buildRegistrationEntriesCache(ctx, log, ds, clk, cache, buildCachePageSize)
+ registrationEntries, err := buildRegistrationEntriesCache(ctx, log, ds, clk, cache, buildCachePageSize, sqlTransactionTimeout)
  if err != nil {
  return nil, nil, nil, err
  }
 
- attestedNodes, err := buildAttestedNodesCache(ctx, log, ds, clk, cache)
+ attestedNodes, err := buildAttestedNodesCache(ctx, log, ds, clk, cache, sqlTransactionTimeout)
  if err != nil {
  return nil, nil, nil, err
  }