Merge pull request #1648 from authzed/backward-compatible-dispatch-si…

…ngleflight

singleflight dispatch: adds a fallback in case the traversal bloom is not present

internal/datastore/proxy/singleflight.go

@@ -12,7 +12,7 @@ import (
 // NewSingleflightDatastoreProxy creates a new Datastore proxy which
 // deduplicates calls to Datastore methods that can share results.
 func NewSingleflightDatastoreProxy(d datastore.Datastore) datastore.Datastore {
-	return &observableProxy{delegate: d}
+	return &singleflightProxy{delegate: d}
 }
 
 type singleflightProxy struct {

internal/dispatch/singleflight/singleflight.go

@@ -3,6 +3,7 @@ package singleflight
 import (
 	"context"
 	"encoding/hex"
+	"fmt"
 	"strconv"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -50,6 +51,20 @@ func (d *Dispatcher) DispatchCheck(ctx context.Context, req *v1.DispatchCheckReq
 
 	keyString := hex.EncodeToString(key)
 
+	// this is in place so that upgrading to a SpiceDB version with traversal bloom does not cause dispatch failures
+	// if this is observed frequently it suggests a callsite is missing setting the bloom filter.
+	// Since there is no bloom filter, there is no guarantee recursion won't happen, so it's safer not to singleflight
+	if len(req.Metadata.TraversalBloom) == 0 {
+		tb, err := v1.NewTraversalBloomFilter(50)
+		if err != nil {
+			return &v1.DispatchCheckResponse{Metadata: &v1.ResponseMeta{DispatchCount: 1}}, status.Error(codes.Internal, fmt.Errorf("unable to create traversal bloom filter: %w", err).Error())
+		}
+
+		singleFlightCount.WithLabelValues("DispatchCheck", "missing").Inc()
+		req.Metadata.TraversalBloom = tb
+		return d.delegate.DispatchCheck(ctx, req)
+	}
+
 	// Check if the key has already been part of a dispatch. If so, this represents a
 	// likely recursive call, so we dispatch it to the delegate to avoid the singleflight from blocking it.
 	// If the bloom filter presents a false positive, a dispatch will happen, which is a small inefficiency
@@ -82,6 +97,21 @@ func (d *Dispatcher) DispatchExpand(ctx context.Context, req *v1.DispatchExpandR
 	}
 
 	keyString := hex.EncodeToString(key)
+
+	// this is in place so that upgrading to a SpiceDB version with traversal bloom does not cause dispatch failures
+	// if this is observed frequently it suggests a callsite is missing setting the bloom filter
+	// Since there is no bloom filter, there is no guarantee recursion won't happen, so it's safer not to singleflight
+	if len(req.Metadata.TraversalBloom) == 0 {
+		tb, err := v1.NewTraversalBloomFilter(50)
+		if err != nil {
+			return &v1.DispatchExpandResponse{Metadata: &v1.ResponseMeta{DispatchCount: 1}}, status.Error(codes.Internal, fmt.Errorf("unable to create traversal bloom filter: %w", err).Error())
+		}
+
+		singleFlightCount.WithLabelValues("DispatchExpand", "missing").Inc()
+		req.Metadata.TraversalBloom = tb
+		return d.delegate.DispatchExpand(ctx, req)
+	}
+
 	possiblyLoop, err := req.Metadata.RecordTraversal(keyString)
 	if err != nil {
 		return &v1.DispatchExpandResponse{Metadata: &v1.ResponseMeta{DispatchCount: 1}}, err

internal/dispatch/singleflight/singleflight_test.go

@@ -256,6 +256,54 @@ func TestSingleFlightDispatcherExpand(t *testing.T) {
 	require.Equal(t, uint64(2), called.Load(), "should have dispatched %d calls but did %d", uint64(2), called.Load())
 }
 
+func TestSingleFlightDispatcherCheckBypassesIfMissingBloomFiler(t *testing.T) {
+	singleFlightCount = prometheus.NewCounterVec(singleFlightCountConfig, []string{"method", "shared"})
+	reg := registerMetricInGatherer(singleFlightCount)
+
+	var called atomic.Uint64
+	f := func() {
+		called.Add(1)
+	}
+	disp := New(mockDispatcher{f: f}, &keys.DirectKeyHandler{})
+
+	req := &v1.DispatchCheckRequest{
+		ResourceRelation: tuple.RelationReference("document", "view"),
+		ResourceIds:      []string{"foo", "bar"},
+		Subject:          tuple.ObjectAndRelation("user", "tom", "..."),
+		Metadata: &v1.ResolverMeta{
+			AtRevision: "1234",
+		},
+	}
+
+	_, _ = disp.DispatchCheck(context.Background(), req.CloneVT())
+
+	require.Equal(t, uint64(1), called.Load(), "should have dispatched %d calls but did %d", uint64(1), called.Load())
+	assertCounterWithLabel(t, reg, 1, "spicedb_dispatch_single_flight_total", "missing")
+}
+
+func TestSingleFlightDispatcherExpandBypassesIfMissingBloomFiler(t *testing.T) {
+	singleFlightCount = prometheus.NewCounterVec(singleFlightCountConfig, []string{"method", "shared"})
+	reg := registerMetricInGatherer(singleFlightCount)
+
+	var called atomic.Uint64
+	f := func() {
+		called.Add(1)
+	}
+	disp := New(mockDispatcher{f: f}, &keys.DirectKeyHandler{})
+
+	req := &v1.DispatchExpandRequest{
+		ResourceAndRelation: tuple.ObjectAndRelation("document", "foo", "view"),
+		Metadata: &v1.ResolverMeta{
+			AtRevision: "1234",
+		},
+	}
+
+	_, _ = disp.DispatchExpand(context.Background(), req.CloneVT())
+
+	require.Equal(t, uint64(1), called.Load(), "should have dispatched %d calls but did %d", uint64(1), called.Load())
+	assertCounterWithLabel(t, reg, 1, "spicedb_dispatch_single_flight_total", "missing")
+}
+
 func registerMetricInGatherer(collector prometheus.Collector) prometheus.Gatherer {
 	reg := prometheus.NewRegistry()
 	reg.MustRegister(collector)