goPublishEvents & OnError updated

pkg/client/events/replay_client.go

@@ -11,6 +11,7 @@ import (
  "github.com/pokt-network/poktroll/pkg/either"
  "github.com/pokt-network/poktroll/pkg/observable"
  "github.com/pokt-network/poktroll/pkg/observable/channel"
+ "github.com/pokt-network/poktroll/pkg/polylog"
  "github.com/pokt-network/poktroll/pkg/retry"
 )
 
@@ -181,24 +182,34 @@ func (rClient *replayClient[T]) Close() {
 // observable returns an asynchronous error.
 // This function is intended to be called in a goroutine.
 func (rClient *replayClient[T]) goPublishEvents(ctx context.Context) {
- // React to errors by getting a new events bytes observable, re-mapping it,
- // and send it to replayObsCachePublishCh such that
- // replayObsCache.Last(ctx, 1) will return it.
- publishError := retry.OnError(
- ctx,
- eventsBytesRetryLimit,
- eventsBytesRetryDelay,
- eventsBytesRetryResetTimeout,
- "goPublishEvents",
- rClient.retryPublishEventsFactory(ctx),
- )
+ logger := polylog.Ctx(ctx)
+
+ for {
+ // React to errors by getting a new events bytes observable, re-mapping it,
+ // and send it to replayObsCachePublishCh such that
+ // replayObsCache.Last(ctx, 1) will return it.
+ publishError := retry.OnError(
+ ctx,
+ eventsBytesRetryLimit,
+ eventsBytesRetryDelay,
+ eventsBytesRetryResetTimeout,
+ "goPublishEvents",
+ rClient.retryPublishEventsFactory(ctx),
+ )
+
+ // OnError returned due to a context close
+ if publishError == nil {
+ return
+ }
 
- // If we get here, the retry limit was reached and the retry loop exited.
- // Since this function runs in a goroutine, we can't return the error to the
- // caller. Instead, we panic.
- if publishError != nil {
- panic(fmt.Errorf("EventsReplayClient[%T].goPublishEvents should never reach this spot: %w", *new(T), publishError))
+ // OnError returned due to an error being received on errCh or errCh being closed
+ // In either case we emit a warning and call start over
+ logger.Warn().
+ Str("work_name", "goPublishEvents").
+ Err(publishError).
+ Msgf("EventsRelayClient[%T].goPublishEvents OnError returned")
  }
+
 }
 
 // retryPublishEventsFactory returns a function which is intended to be passed

pkg/retry/retry.go

@@ -2,6 +2,7 @@ package retry
 
 import (
  "context"
+ "errors"
  "time"
 
  "github.com/pokt-network/poktroll/pkg/polylog"
@@ -55,7 +56,7 @@ func OnError(
  logger.Warn().
  Str("work_name", workName).
  Msg("error channel closed, will no longer retry on error")
- return nil
+ return errors.New("error channel closed")
  }
 
  if retryLimit > 0 && retryCount >= retryLimit {