Add comments to events based cache code (#5327)

* Add comments to events based cache
* Remove scrolling debug log

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

pkg/server/endpoints/authorized_entryfetcher_attested_nodes.go

@@ -31,7 +31,8 @@ type attestedNodes struct {
  sqlTransactionTimeout time.Duration
 }
 
-// buildAttestedNodesCache Fetches all attested nodes and adds the unexpired ones to the cache
+// buildAttestedNodesCache fetches all attested nodes and adds the unexpired ones to the cache.
+// It runs once at startup.
 func buildAttestedNodesCache(ctx context.Context, log logrus.FieldLogger, ds datastore.DataStore, clk clock.Clock, cache *authorizedentries.Cache, sqlTransactionTimeout time.Duration) (*attestedNodes, error) {
  resp, err := ds.ListAttestedNodesEvents(ctx, &datastore.ListAttestedNodesEventsRequest{})
  if err != nil {
@@ -49,6 +50,8 @@ func buildAttestedNodesCache(ctx context.Context, log logrus.FieldLogger, ds dat
  firstEventID = event.EventID
  firstEventTime = now
  } else {
+ // After getting the first event, search for any gaps in the event stream, from the first event to the last event.
+ // During each cache refresh cycle, we will check if any of these missed events get populated.
  for i := lastEventID + 1; i < event.EventID; i++ {
  missedEvents[i] = now
  }
@@ -89,11 +92,10 @@ func buildAttestedNodesCache(ctx context.Context, log logrus.FieldLogger, ds dat
 // updateCache Fetches all the events since the last time this function was running and updates
 // the cache with all the changes.
 func (a *attestedNodes) updateCache(ctx context.Context) error {
+ // Process events skipped over previously
  if err := a.missedStartupEvents(ctx); err != nil {
  a.log.WithError(err).Error("Unable to process missed startup events")
  }
-
- // Process events skipped over previously
  a.replayMissedEvents(ctx)
 
  req := &datastore.ListAttestedNodesEventsRequest{
@@ -111,8 +113,8 @@ func (a *attestedNodes) updateCache(ctx context.Context) error {
  // were skipped over and need to be queued in case they show up later.
  // This can happen when a long running transaction allocates an event ID but a shorter transaction
  // comes in after, allocates and commits the ID first. If a read comes in at this moment, the event id for
- // the longer running transaction will be skipped over
- if !a.firstEventTime.IsZero() && event.EventID != a.lastEventID+1 {
+ // the longer running transaction will be skipped over.
+ if !a.firstEventTime.IsZero() {
  for i := a.lastEventID + 1; i < event.EventID; i++ {
  a.log.WithField(telemetry.EventID, i).Info("Detected skipped attested node event")
  a.mu.Lock()
@@ -143,6 +145,10 @@ func (a *attestedNodes) updateCache(ctx context.Context) error {
  return nil
 }
 
+// missedStartupEvents will check for any events that arrive with an ID less than the first event ID we receive.
+// For example if the first event ID we receive is 3, this function will check for any IDs less than that.
+// If event ID 2 comes in later on, due to a long running transaction, this function will update the cache
+// with the information from this event. This function will run until time equal to sqlTransactionTimeout has elapsed after startup.
 func (a *attestedNodes) missedStartupEvents(ctx context.Context) error {
  if a.firstEventTime.IsZero() || a.clk.Now().Sub(a.firstEventTime) > a.sqlTransactionTimeout {
  return nil
@@ -182,7 +188,6 @@ func (a *attestedNodes) replayMissedEvents(ctx context.Context) {
  switch status.Code(err) {
  case codes.OK:
  case codes.NotFound:
- log.Debug("Event not yet populated in database")
  continue
  default:
  log.WithError(err).Error("Failed to fetch info about missed Attested Node event")

pkg/server/endpoints/authorized_entryfetcher_attested_nodes_test.go

@@ -95,7 +95,7 @@ func TestAttestedNodesCacheMissedEventNotFound(t *testing.T) {
 
  attestedNodes.missedEvents[1] = clk.Now()
  attestedNodes.replayMissedEvents(ctx)
- require.Equal(t, "Event not yet populated in database", hook.LastEntry().Message)
+ require.Zero(t, hook.Entries)
 }
 
 func TestAttestedNodesSavesMissedStartupEvents(t *testing.T) {

pkg/server/endpoints/authorized_entryfetcher_registration_entries.go

@@ -31,7 +31,8 @@ type registrationEntries struct {
  sqlTransactionTimeout time.Duration
 }
 
-// buildRegistrationEntriesCache Fetches all registration entries and adds them to the cache
+// buildRegistrationEntriesCache fetches all registration entries and adds them to the cache.
+// It runs once at startup.
 func buildRegistrationEntriesCache(ctx context.Context, log logrus.FieldLogger, ds datastore.DataStore, clk clock.Clock, cache *authorizedentries.Cache, pageSize int32, sqlTransactionTimeout time.Duration) (*registrationEntries, error) {
  resp, err := ds.ListRegistrationEntriesEvents(ctx, &datastore.ListRegistrationEntriesEventsRequest{})
  if err != nil {
@@ -49,6 +50,8 @@ func buildRegistrationEntriesCache(ctx context.Context, log logrus.FieldLogger,
  firstEventID = event.EventID
  firstEventTime = now
  } else {
+ // After getting the first event, search for any gaps in the event stream, from the first event to the last event.
+ // During each cache refresh cycle, we will check if any of these missed events get populated.
  for i := lastEventID + 1; i < event.EventID; i++ {
  missedEvents[i] = clk.Now()
  }
@@ -102,11 +105,10 @@ func buildRegistrationEntriesCache(ctx context.Context, log logrus.FieldLogger,
 // updateCache Fetches all the events since the last time this function was running and updates
 // the cache with all the changes.
 func (a *registrationEntries) updateCache(ctx context.Context) error {
+ // Process events skipped over previously
  if err := a.missedStartupEvents(ctx); err != nil {
  a.log.WithError(err).Error("Unable to process missed startup events")
  }
-
- // Process events skipped over previously
  a.replayMissedEvents(ctx)
 
  req := &datastore.ListRegistrationEntriesEventsRequest{
@@ -124,8 +126,8 @@ func (a *registrationEntries) updateCache(ctx context.Context) error {
  // were skipped over and need to be queued in case they show up later.
  // This can happen when a long running transaction allocates an event ID but a shorter transaction
  // comes in after, allocates and commits the ID first. If a read comes in at this moment, the event id for
- // the longer running transaction will be skipped over
- if !a.firstEventTime.IsZero() && event.EventID != a.lastEventID+1 {
+ // the longer running transaction will be skipped over.
+ if !a.firstEventTime.IsZero() {
  for i := a.lastEventID + 1; i < event.EventID; i++ {
  a.log.WithField(telemetry.EventID, i).Info("Detected skipped registration entry event")
  a.mu.Lock()
@@ -156,6 +158,10 @@ func (a *registrationEntries) updateCache(ctx context.Context) error {
  return nil
 }
 
+// missedStartupEvents will check for any events come in with an ID less than the first event ID we receive.
+// For example if the first event ID we receive is 3, this function will check for any IDs less than that.
+// If event ID 2 comes in later on, due to a long running transaction, this function will update the cache
+// with the information from this event. This function will run until time equal to sqlTransactionTimeout has elapsed after startup.
 func (a *registrationEntries) missedStartupEvents(ctx context.Context) error {
  if a.firstEventTime.IsZero() || a.clk.Now().Sub(a.firstEventTime) > a.sqlTransactionTimeout {
  return nil
@@ -195,7 +201,6 @@ func (a *registrationEntries) replayMissedEvents(ctx context.Context) {
  switch status.Code(err) {
  case codes.OK:
  case codes.NotFound:
- log.Debug("Event not yet populated in database")
  continue
  default:
  log.WithError(err).Error("Failed to fetch info about missed event")

pkg/server/endpoints/authorized_entryfetcher_registration_entries_test.go

@@ -111,7 +111,7 @@ func TestRegistrationEntriesCacheMissedEventNotFound(t *testing.T) {
 
  registrationEntries.missedEvents[1] = clk.Now()
  registrationEntries.replayMissedEvents(ctx)
- require.Equal(t, "Event not yet populated in database", hook.LastEntry().Message)
+ require.Zero(t, len(hook.Entries))
 }
 
 func TestRegistrationEntriesSavesMissedStartupEvents(t *testing.T) {