Add API support for transaction metadata on WriteRels and DeleteRels

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/IBM/pgxpoolprometheus v1.1.1
 	github.com/KimMachineGun/automemlimit v0.6.1
 	github.com/Masterminds/squirrel v1.5.4
-	github.com/authzed/authzed-go v0.15.1-0.20240916185322-dad4080470f3
+	github.com/authzed/authzed-go v0.16.1-0.20241001202507-27cc182a7b92
 
 	// NOTE: We are using a *copy* of `cel-go` here to ensure there isn't a conflict
 	// with the version used in Kubernetes. This is a temporary measure until we can

go.sum

@@ -702,8 +702,8 @@ github.com/ashanbrown/forbidigo v1.6.0 h1:D3aewfM37Yb3pxHujIPSpTf6oQk9sc9WZi8ger
 github.com/ashanbrown/forbidigo v1.6.0/go.mod h1:Y8j9jy9ZYAEHXdu723cUlraTqbzjKF1MUyfOKL+AjcU=
 github.com/ashanbrown/makezero v1.1.1 h1:iCQ87C0V0vSyO+M9E/FZYbu65auqH0lnsOkf5FcB28s=
 github.com/ashanbrown/makezero v1.1.1/go.mod h1:i1bJLCRSCHOcOa9Y6MyF2FTfMZMFdHvxKHxgO5Z1axI=
-github.com/authzed/authzed-go v0.15.1-0.20240916185322-dad4080470f3 h1:M8rXwNE9iTtNiIs/OgEXhwOQmGa1MLYv4+WOiOjixgQ=
-github.com/authzed/authzed-go v0.15.1-0.20240916185322-dad4080470f3/go.mod h1:Cx1DQKMX38u2fFVLZiGUuZnbTo2J7LCZEXmVak+TMak=
+github.com/authzed/authzed-go v0.16.1-0.20241001202507-27cc182a7b92 h1:9BQQO4ga8naS5bl2aWeRoyF54Bxp1H28WMr+/XU7rLE=
+github.com/authzed/authzed-go v0.16.1-0.20241001202507-27cc182a7b92/go.mod h1:Cx1DQKMX38u2fFVLZiGUuZnbTo2J7LCZEXmVak+TMak=
 github.com/authzed/cel-go v0.20.2 h1:GlmLecGry7Z8HU0k+hmaHHUV05ZHrsFxduXHtIePvck=
 github.com/authzed/cel-go v0.20.2/go.mod h1:pJHVFWbqUHV1J+klQoZubdKswlbxcsbojda3mye9kiU=
 github.com/authzed/consistent v0.1.0 h1:tlh1wvKoRbjRhMm2P+X5WQQyR54SRoS4MyjLOg17Mp8=

internal/services/v1/errors.go

@@ -475,3 +475,37 @@ func defaultIfZero[T comparable](value T, defaultValue T) T {
 	}
 	return value
 }
+
+// ErrTransactionMetadataTooLarge indicates that the metadata for a transaction is too large.
+type ErrTransactionMetadataTooLarge struct {
+	error
+	metadataSize int
+	maxSize      int
+}
+
+// NewTransactionMetadataTooLargeErr constructs a new transaction metadata too large error.
+func NewTransactionMetadataTooLargeErr(metadataSize int, maxSize int) ErrTransactionMetadataTooLarge {
+	return ErrTransactionMetadataTooLarge{
+		error:        fmt.Errorf("metadata size of %d is greater than maximum allowed of %d", metadataSize, maxSize),
+		metadataSize: metadataSize,
+		maxSize:      maxSize,
+	}
+}
+
+func (err ErrTransactionMetadataTooLarge) MarshalZerologObject(e *zerolog.Event) {
+	e.Err(err.error).Int("metadataSize", err.metadataSize).Int("maxSize", err.maxSize)
+}
+
+func (err ErrTransactionMetadataTooLarge) GRPCStatus() *status.Status {
+	return spiceerrors.WithCodeAndDetails(
+		err,
+		codes.InvalidArgument,
+		spiceerrors.ForReason(
+			v1.ErrorReason_ERROR_REASON_TRANSACTION_METADATA_TOO_LARGE,
+			map[string]string{
+				"metadata_byte_size":                 strconv.Itoa(err.metadataSize),
+				"maximum_allowed_metadata_byte_size": strconv.Itoa(err.maxSize),
+			},
+		),
+	)
+}

internal/services/v1/relationships.go

@@ -12,6 +12,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/authzed/spicedb/internal/dispatch"
 	"github.com/authzed/spicedb/internal/middleware"
@@ -273,6 +274,10 @@ func (ps *permissionServer) ReadRelationships(req *v1.ReadRelationshipsRequest,
 }
 
 func (ps *permissionServer) WriteRelationships(ctx context.Context, req *v1.WriteRelationshipsRequest) (*v1.WriteRelationshipsResponse, error) {
+	if err := ps.validateTransactionMetadata(req.OptionalTransactionMetadata); err != nil {
+		return nil, ps.rewriteError(ctx, err)
+	}
+
 	ds := datastoremw.MustFromContext(ctx)
 
 	span := trace.SpanFromContext(ctx)
@@ -347,7 +352,7 @@ func (ps *permissionServer) WriteRelationships(ctx context.Context, req *v1.Writ
 
 		span.AddEvent("write relationships")
 		return rwt.WriteRelationships(ctx, tupleUpdates)
-	})
+	}, options.WithMetadata(req.OptionalTransactionMetadata))
 	if err != nil {
 		return nil, ps.rewriteError(ctx, err)
 	}
@@ -367,7 +372,30 @@ func (ps *permissionServer) WriteRelationships(ctx context.Context, req *v1.Writ
 	}, nil
 }
 
+const MaximumTransactionMetadataSize = 65000 // bytes
+
+func (ps *permissionServer) validateTransactionMetadata(metadata *structpb.Struct) error {
+	if metadata == nil {
+		return nil
+	}
+
+	b, err := metadata.MarshalJSON()
+	if err != nil {
+		return err
+	}
+
+	if len(b) > MaximumTransactionMetadataSize {
+		return NewTransactionMetadataTooLargeErr(len(b), MaximumTransactionMetadataSize)
+	}
+
+	return nil
+}
+
 func (ps *permissionServer) DeleteRelationships(ctx context.Context, req *v1.DeleteRelationshipsRequest) (*v1.DeleteRelationshipsResponse, error) {
+	if err := ps.validateTransactionMetadata(req.OptionalTransactionMetadata); err != nil {
+		return nil, ps.rewriteError(ctx, err)
+	}
+
 	if len(req.OptionalPreconditions) > int(ps.config.MaxPreconditionsCount) {
 		return nil, ps.rewriteError(
 			ctx,
@@ -456,7 +484,7 @@ func (ps *permissionServer) DeleteRelationships(ctx context.Context, req *v1.Del
 		// Otherwise, kick off an unlimited deletion.
 		_, err = rwt.DeleteRelationships(ctx, req.RelationshipFilter)
 		return err
-	})
+	}, options.WithMetadata(req.OptionalTransactionMetadata))
 	if err != nil {
 		return nil, ps.rewriteError(ctx, err)
 	}

internal/services/v1/relationships_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"maps"
+	"strings"
 	"testing"
 	"time"
 
@@ -1344,6 +1345,113 @@ func TestDeleteRelationshipsPreconditionsOverLimit(t *testing.T) {
 	require.Contains(err.Error(), "precondition count of 2 is greater than maximum allowed of 1")
 }
 
+func TestWriteRelationshipsWithMetadata(t *testing.T) {
+	require := require.New(t)
+	conn, cleanup, _, beforeWriteRev := testserver.NewTestServerWithConfig(
+		require,
+		testTimedeltas[0],
+		memdb.DisableGC,
+		true,
+		testserver.ServerConfig{
+			MaxPreconditionsCount: 10,
+			MaxUpdatesPerWrite:    10,
+		},
+		tf.StandardDatastoreWithData,
+	)
+	t.Cleanup(cleanup)
+
+	client := v1.NewPermissionsServiceClient(conn)
+
+	metadata, err := structpb.NewStruct(map[string]any{
+		"foo": "123546",
+	})
+	require.NoError(err)
+
+	_, err = client.WriteRelationships(context.Background(), &v1.WriteRelationshipsRequest{
+		OptionalTransactionMetadata: metadata,
+		Updates: []*v1.RelationshipUpdate{
+			{
+				Operation:    v1.RelationshipUpdate_OPERATION_TOUCH,
+				Relationship: rel("document", "newdoc", "parent", "folder", "afolder", ""),
+			},
+		},
+	})
+
+	require.NoError(err)
+
+	beforeWriteToken := zedtoken.MustNewFromRevision(beforeWriteRev)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	watchClient := v1.NewWatchServiceClient(conn)
+
+	stream, err := watchClient.Watch(ctx, &v1.WatchRequest{OptionalStartCursor: beforeWriteToken})
+	require.NoError(err)
+
+	resp, err := stream.Recv()
+	require.NoError(err)
+	require.Equal(metadata, resp.OptionalTransactionMetadata)
+}
+
+func TestWriteRelationshipsMetadataOverLimit(t *testing.T) {
+	require := require.New(t)
+	conn, cleanup, _, _ := testserver.NewTestServerWithConfig(
+		require,
+		testTimedeltas[0],
+		memdb.DisableGC,
+		true,
+		testserver.ServerConfig{
+			MaxPreconditionsCount: 1,
+			MaxUpdatesPerWrite:    1,
+		},
+		tf.StandardDatastoreWithData,
+	)
+	client := v1.NewPermissionsServiceClient(conn)
+	t.Cleanup(cleanup)
+
+	metadata, err := structpb.NewStruct(map[string]any{
+		"foo": strings.Repeat("hithere", 65000),
+	})
+	require.NoError(err)
+
+	_, err = client.WriteRelationships(context.Background(), &v1.WriteRelationshipsRequest{
+		OptionalTransactionMetadata: metadata,
+	})
+
+	require.Error(err)
+	require.Contains(err.Error(), "metadata size of 455010 is greater than maximum allowed of 65000")
+}
+
+func TestDeleteRelationshipsMetadataOverLimit(t *testing.T) {
+	require := require.New(t)
+	conn, cleanup, _, _ := testserver.NewTestServerWithConfig(
+		require,
+		testTimedeltas[0],
+		memdb.DisableGC,
+		true,
+		testserver.ServerConfig{
+			MaxPreconditionsCount: 1,
+			MaxUpdatesPerWrite:    1,
+		},
+		tf.StandardDatastoreWithData,
+	)
+	client := v1.NewPermissionsServiceClient(conn)
+	t.Cleanup(cleanup)
+
+	metadata, err := structpb.NewStruct(map[string]any{
+		"foo": strings.Repeat("hithere", 65000),
+	})
+	require.NoError(err)
+
+	_, err = client.DeleteRelationships(context.Background(), &v1.DeleteRelationshipsRequest{
+		OptionalTransactionMetadata: metadata,
+		RelationshipFilter:          &v1.RelationshipFilter{},
+	})
+
+	require.Error(err)
+	require.Contains(err.Error(), "metadata size of 455010 is greater than maximum allowed of 65000")
+}
+
 func TestWriteRelationshipsPreconditionsOverLimit(t *testing.T) {
 	require := require.New(t)
 	conn, cleanup, _, _ := testserver.NewTestServerWithConfig(

internal/services/v1/watch.go

@@ -96,8 +96,9 @@ func (ws *watchServer) Watch(req *v1.WatchRequest, stream v1.WatchService_WatchS
 				filtered := filterUpdates(objectTypes, filters, update.RelationshipChanges)
 				if len(filtered) > 0 {
 					if err := stream.Send(&v1.WatchResponse{
-						Updates:        filtered,
-						ChangesThrough: zedtoken.MustNewFromRevision(update.Revision),
+						Updates:                     filtered,
+						ChangesThrough:              zedtoken.MustNewFromRevision(update.Revision),
+						OptionalTransactionMetadata: update.Metadata,
 					}); err != nil {
 						return status.Errorf(codes.Canceled, "watch canceled by user: %s", err)
 					}