More specific pushback errors (#782)

* Add reason field to pushback errors

* simplify everything

* even simpler

* inside

* comments

Author: phbnf

append_lifecycle.go

@@ -20,7 +20,6 @@ import (
 	"fmt"
 	"net/http"
 	"os"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -407,20 +406,23 @@ func (i *integrationStats) statsDecorator(delegate AddFn) AddFn {
 		return func() (Index, error) {
 			idx, err := f()
 			attr := []attribute.KeyValue{}
-			pushbackType := "" // This will be used for the pushback attribute below, empty string means no pushback
 
 			if err != nil {
-				if errors.Is(err, ErrPushback) {
-					// record the the fact there was pushback, and use the error string as the type.
-					pushbackType = err.Error()
-				} else {
-					// Just flag that it's an errored request to avoid high cardinality of attribute values.
+				switch {
+				// Record the fact there was pushback, if any.
+				case errors.Is(err, ErrPushbackAntispam):
+					attr = append(attr, attribute.String("tessera.pushback", "antispam"))
+				case errors.Is(err, ErrPushbackIntegration):
+					attr = append(attr, attribute.String("tessera.pushback", "integration"))
+				case errors.Is(err, ErrPushback):
+					attr = append(attr, attribute.String("tessera.pushback", "other"))
+				default:
+					// If it's not a pushback, just flag that it's an errored request to avoid high cardinality of attribute values.
 					// TODO(al): We might want to bucket errors into OTel status codes in the future, though.
 					attr = append(attr, attribute.String("tessera.error.type", "_OTHER"))
 				}
 			}
 
-			attr = append(attr, attribute.String("tessera.pushback", strings.ReplaceAll(pushbackType, " ", "_")))
 			attr = append(attr, attribute.Bool("tessera.duplicate", idx.IsDup))
 
 			appenderAddsTotal.Add(ctx, 1, metric.WithAttributes(attr...))

log.go

@@ -16,18 +16,30 @@ package tessera
 
 import (
 	"errors"
+	"fmt"
 )
 
-// ErrPushback is returned by underlying storage implementations when a new entry cannot be accepted
-// due to overload in the system. This could be because there are too many entries with indices assigned
-// but which have not yet been integrated into the tree, or it could be because the antispam mechanism
-// is not able to keep up with recently added entries.
-//
-// Personalities encountering this error should apply back-pressure to the source of new entries
-// in an appropriate manner (e.g. for HTTP services, return a 503 with a Retry-After header).
-//
-// Personalities should check for this error using `errors.Is(e, ErrPushback)`.
-var ErrPushback = errors.New("pushback")
+var (
+	// ErrPushback is returned by underlying storage implementations when a new entry cannot be accepted
+	// due to overload in the system. This could be because there are too many entries with indices assigned
+	// but which have not yet been integrated into the tree, or it could be because the antispam mechanism
+	// is not able to keep up with recently added entries. It should always be wrapped with a more
+	// specific error to provide context to clients.
+	//
+	// Personalities encountering this error should apply back-pressure to the source of new entries
+	// in an appropriate manner (e.g. for HTTP services, return a 503 with a Retry-After header).
+	//
+	// Personalities should check for this error (wrapped or not) using `errors.Is(e, ErrPushback)`.
+	ErrPushback = errors.New("pushback")
+	// ErrPushbackAntispam is a wrapped ErrPushback. It is returned by underlying storage implementations
+	// when an entry cannot be accepted becasue the antispam follower has fallen too far behind the size
+	// of the integrated tree.
+	ErrPushbackAntispam = fmt.Errorf("antispam %w", ErrPushback)
+	// ErrPushbackIntegration is a wrapped ErrPushback. It is returned by underlying storage implementations
+	// when an entry cannot be accepted becasue there are too many "in-flight" add requests - i.e. entries
+	// with sequence numbers assigned, but which are not yet integrated into the log.
+	ErrPushbackIntegration = fmt.Errorf("integration %w", ErrPushback)
+)
 
 // Driver is the implementation-specific parts of Tessera. No methods are on here as this is not for public use.
 type Driver any

storage/aws/antispam/aws.go

@@ -49,8 +49,6 @@ const (
 	SchemaCompatibilityVersion = 1
 )
 
-var errPushback = fmt.Errorf("antispam %w", tessera.ErrPushback)
-
 // AntispamOpts allows configuration of some tunable options.
 type AntispamOpts struct {
 	// MaxBatchSize is the largest number of mutations permitted in a single write operation when
@@ -215,7 +213,7 @@ func (d *AntispamStorage) Decorator() func(f tessera.AddFn) tessera.AddFn {
 				//
 				// We may decide in the future that serving duplicate reads is more important than catching up as quickly
 				// as possible, in which case we'd move this check down below the call to index.
-				return func() (tessera.Index, error) { return tessera.Index{}, errPushback }
+				return func() (tessera.Index, error) { return tessera.Index{}, tessera.ErrPushbackAntispam }
 			}
 			idx, err := d.index(ctx, e.Identity())
 			if err != nil {

storage/aws/aws.go

@@ -1021,7 +1021,7 @@ func (s *mySQLSequencer) assignEntries(ctx context.Context, entries []*tessera.E
 	// Check whether there are too many outstanding entries and we should apply
 	// back-pressure.
 	if outstanding := next - treeSize; outstanding > s.maxOutstanding {
-		return tessera.ErrPushback
+		return tessera.ErrPushbackIntegration
 	}
 
 	sequencedEntries := make([]storage.SequencedEntry, len(entries))

storage/gcp/antispam/gcp.go

@@ -47,8 +47,6 @@ const (
 	DefaultPushbackThreshold = 2048
 )
 
-var errPushback = fmt.Errorf("antispam %w", tessera.ErrPushback)
-
 // AntispamOpts allows configuration of some tunable options.
 type AntispamOpts struct {
 	// MaxBatchSize is the largest number of mutations permitted in a single BatchWrite operation when
@@ -168,7 +166,7 @@ func (d *AntispamStorage) Decorator() func(f tessera.AddFn) tessera.AddFn {
 				//
 				// We may decide in the future that serving duplicate reads is more important than catching up as quickly
 				// as possible, in which case we'd move this check down below the call to index.
-				return func() (tessera.Index, error) { return tessera.Index{}, errPushback }
+				return func() (tessera.Index, error) { return tessera.Index{}, tessera.ErrPushbackAntispam }
 			}
 			idx, err := d.index(ctx, e.Identity())
 			if err != nil {

storage/gcp/gcp.go

@@ -808,7 +808,7 @@ func (s *spannerCoordinator) assignEntries(ctx context.Context, entries []*tesse
 		// Check whether there are too many outstanding entries and we should apply
 		// back-pressure.
 		if outstanding := next - treeSize; outstanding > int64(s.maxOutstanding) {
-			return tessera.ErrPushback
+			return tessera.ErrPushbackIntegration
 		}
 
 		next := uint64(next) // Shadow next with a uint64 version of the same value to save on casts.

storage/posix/antispam/badger.go

@@ -170,7 +170,7 @@ func (d *AntispamStorage) Decorator() func(f tessera.AddFn) tessera.AddFn {
 				//
 				// We may decide in the future that serving duplicate reads is more important than catching up as quickly
 				// as possible, in which case we'd move this check down below the call to index.
-				return func() (tessera.Index, error) { return tessera.Index{}, tessera.ErrPushback }
+				return func() (tessera.Index, error) { return tessera.Index{}, tessera.ErrPushbackAntispam }
 			}
 			idx, err := d.index(ctx, e.Identity())
 			if err != nil {