fix: data races that occur when functions do not clone pointer object… (#2629)

Author: miparnisari

internal/dispatch/caching/caching.go

@@ -196,7 +196,7 @@ func (cd *Dispatcher) DispatchCheck(ctx context.Context, req *v1.DispatchCheckRe
 
 	// Return both the computed and err in ALL cases: computed contains resolved
 	// metadata even if there was an error.
-	return computed, err
+	return computed.CloneVT(), err
 }
 
 // DispatchExpand implements dispatch.Expand interface and does not do any caching yet.

internal/dispatch/caching/cachingdispatch_test.go

@@ -2,6 +2,7 @@ package caching
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 
@@ -149,6 +150,77 @@ func TestMaxDepthCaching(t *testing.T) {
 	}
 }
 
+func TestConcurrentDebugInfoAccess(t *testing.T) {
+	require := require.New(t)
+
+	delegate := delegateDispatchMock{&mock.Mock{}}
+
+	originalReq := &v1.DispatchCheckRequest{
+		ResourceIds: []string{"original"},
+		Subject: &core.ObjectAndRelation{
+			Relation: "original",
+		},
+	}
+
+	// Have the delegate return one request object for all calls
+	delegate.On("DispatchCheck", mock.Anything).
+		Return(&v1.DispatchCheckResponse{
+			Metadata: &v1.ResponseMeta{
+				DispatchCount: 1,
+				DebugInfo:     &v1.DebugInformation{Check: &v1.CheckDebugTrace{Request: originalReq}},
+			},
+		}, nil)
+
+	dispatcher, err := NewCachingDispatcher(DispatchTestCache(t), false, "", nil)
+	require.NoError(err)
+	dispatcher.SetDelegate(delegate)
+	t.Cleanup(func() {
+		_ = dispatcher.Close()
+	})
+
+	// Spawn multiple goroutines that create requests with slice views of the same backing array
+	// This simulates what happens in ComputeBulkCheck with ForEachChunkUntil
+	const numGoroutines = 1000
+	errors := make(chan error, numGoroutines)
+
+	var wg sync.WaitGroup
+	for i := 0; i < numGoroutines; i++ {
+		wg.Add(1)
+		go func(goroutineID int) {
+			defer wg.Done()
+
+			request := &v1.DispatchCheckRequest{
+				ResourceRelation: RR("document", "viewer"),
+				ResourceIds:      []string{"doc1"},
+				Subject:          tuple.MustParseSubjectONR("user:alice").ToCoreONR(),
+				Metadata: &v1.ResolverMeta{
+					AtRevision:     decimal.Zero.String(),
+					DepthRemaining: 50,
+				},
+				Debug: v1.DispatchCheckRequest_ENABLE_BASIC_DEBUGGING,
+			}
+
+			resp, err := dispatcher.DispatchCheck(context.Background(), request)
+			if err != nil {
+				errors <- err
+				return
+			}
+
+			require.NotNil(resp.GetMetadata().GetDebugInfo().GetCheck().GetRequest())
+
+			// we mutate the response to prove that it's not shared across goroutines
+			resp.GetMetadata().GetDebugInfo().GetCheck().GetRequest().Subject.Relation = "modified"
+			resp.GetMetadata().GetDebugInfo().GetCheck().GetRequest().ResourceIds = []string{"modified"}
+		}(i)
+	}
+
+	wg.Wait()
+	close(errors)
+	for err := range errors {
+		require.NoError(err)
+	}
+}
+
 type delegateDispatchMock struct {
 	*mock.Mock
 }

internal/dispatch/dispatch.go

@@ -37,6 +37,7 @@ type Dispatcher interface {
 // Check interface describes just the methods required to dispatch check requests.
 type Check interface {
 	// DispatchCheck submits a single check request and returns its result.
+	// The result should be safe to access from multiple goroutines.
 	DispatchCheck(ctx context.Context, req *v1.DispatchCheckRequest) (*v1.DispatchCheckResponse, error)
 }
 

internal/dispatch/singleflight/singleflight.go

@@ -88,7 +88,7 @@ func (d *Dispatcher) DispatchCheck(ctx context.Context, req *v1.DispatchCheckReq
 	}
 
 	singleFlightCount.WithLabelValues("DispatchCheck", strconv.FormatBool(isShared)).Inc()
-	return sharedResp, err
+	return sharedResp.CloneVT(), err
 }
 
 func (d *Dispatcher) DispatchExpand(ctx context.Context, req *v1.DispatchExpandRequest) (*v1.DispatchExpandResponse, error) {

internal/dispatch/singleflight/singleflight_test.go

@@ -42,21 +42,28 @@ func TestSingleFlightDispatcher(t *testing.T) {
 	wg := sync.WaitGroup{}
 	wg.Add(4)
 	go func() {
-		_, _ = disp.DispatchCheck(t.Context(), req.CloneVT())
+		resp1, err := disp.DispatchCheck(t.Context(), req.CloneVT())
+		require.NoError(t, err)
+		// this goroutine mutates the response; other goroutines that read should be unaffected
+		resp1.GetMetadata().GetDebugInfo().GetCheck().GetSubProblems()[0].IsCachedResult = false
 		wg.Done()
 	}()
 	go func() {
-		_, _ = disp.DispatchCheck(t.Context(), req.CloneVT())
+		resp2, _ := disp.DispatchCheck(t.Context(), req.CloneVT())
+		// this goroutine reads the response
+		t.Log(resp2)
 		wg.Done()
 	}()
 	go func() {
-		_, _ = disp.DispatchCheck(t.Context(), req.CloneVT())
+		resp3, _ := disp.DispatchCheck(t.Context(), req.CloneVT())
+		t.Log(resp3)
 		wg.Done()
 	}()
 	go func() {
 		anotherReq := req.CloneVT()
 		anotherReq.ResourceIds = []string{"foo", "baz"}
-		_, _ = disp.DispatchCheck(t.Context(), anotherReq)
+		resp4, _ := disp.DispatchCheck(t.Context(), anotherReq)
+		t.Log(resp4)
 		wg.Done()
 	}()
 
@@ -363,7 +370,9 @@ type mockDispatcher struct {
 
 func (m mockDispatcher) DispatchCheck(_ context.Context, _ *v1.DispatchCheckRequest) (*v1.DispatchCheckResponse, error) {
 	m.f()
-	return &v1.DispatchCheckResponse{}, nil
+	return &v1.DispatchCheckResponse{Metadata: &v1.ResponseMeta{DebugInfo: &v1.DebugInformation{Check: &v1.CheckDebugTrace{
+		SubProblems: []*v1.CheckDebugTrace{{IsCachedResult: true}},
+	}}}}, nil
 }
 
 func (m mockDispatcher) DispatchExpand(_ context.Context, _ *v1.DispatchExpandRequest) (*v1.DispatchExpandResponse, error) {