feat: batched and chunked insertion+deletion of relation tuples (#1631)

.github/workflows/ci.yaml

@@ -49,7 +49,7 @@ jobs:
     runs-on: ubuntu-latest
     services:
       postgres:
-        image: postgres:11.8
+        image: postgres:16
         env:
           POSTGRES_DB: keto
           POSTGRES_PASSWORD: test
@@ -69,7 +69,7 @@ jobs:
     steps:
       - run: |
           docker create --name cockroach -p 26257:26257 \
-            cockroachdb/cockroach:latest-v23.2 start-single-node --insecure
+            cockroachdb/cockroach:latest-v24.2 start-single-node --insecure
           docker start cockroach
         name: Start CockroachDB
       - uses: ory/ci/checkout@master

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/ory/herodot v0.10.3-0.20230626083119-d7e5192f0d88
 	github.com/ory/jsonschema/v3 v3.0.8
 	github.com/ory/keto/proto v0.13.0-alpha.0
-	github.com/ory/x v0.0.675
+	github.com/ory/x v0.0.677
 	github.com/pelletier/go-toml v1.9.5
 	github.com/phayes/freeport v0.0.0-20220201140144-74d24b5ae9f5
 	github.com/pkg/errors v0.9.1

go.sum

@@ -419,8 +419,8 @@ github.com/ory/jsonschema/v3 v3.0.8 h1:Ssdb3eJ4lDZ/+XnGkvQS/te0p+EkolqwTsDOCxr/F
 github.com/ory/jsonschema/v3 v3.0.8/go.mod h1:ZPzqjDkwd3QTnb2Z6PAS+OTvBE2x5i6m25wCGx54W/0=
 github.com/ory/pop/v6 v6.2.1-0.20241121111754-e5dfc0f3344b h1:BIzoOe2/wynZBQak1po0tzgvARseIKsR2bF6b+SZoKE=
 github.com/ory/pop/v6 v6.2.1-0.20241121111754-e5dfc0f3344b/go.mod h1:okVAYKGtgunD/wbW3NGhZTndJCS+6FqO+cA89rQ4doc=
-github.com/ory/x v0.0.675 h1:K6GpVo99BXBFv2UiwMjySNNNqCFKGswynrt7vWQJFU8=
-github.com/ory/x v0.0.675/go.mod h1:zJmnDtKje2FCP4EeFvRsKk94XXiqKCSGJMZcirAfhUs=
+github.com/ory/x v0.0.677 h1:ZulzE4EBhNBXNotWmGSmGsVNbgbZpIr4snMURRkski0=
+github.com/ory/x v0.0.677/go.mod h1:zJmnDtKje2FCP4EeFvRsKk94XXiqKCSGJMZcirAfhUs=
 github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8=
 github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

internal/e2e/cli_client_test.go

@@ -12,47 +12,37 @@ import (
 	"testing"
 	"time"
 
-	"github.com/ory/keto/ketoapi"
-
 	"github.com/ory/herodot"
-
-	"github.com/ory/keto/internal/check"
-
-	grpcHealthV1 "google.golang.org/grpc/health/grpc_health_v1"
-
-	"github.com/ory/keto/internal/x"
-
+	"github.com/ory/x/cmdx"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	grpcHealthV1 "google.golang.org/grpc/health/grpc_health_v1"
 
 	gprclient "github.com/ory/keto/cmd/client"
 	cliexpand "github.com/ory/keto/cmd/expand"
 	clirelationtuple "github.com/ory/keto/cmd/relationtuple"
-
-	"github.com/ory/x/cmdx"
+	"github.com/ory/keto/internal/check"
+	"github.com/ory/keto/internal/x"
+	"github.com/ory/keto/ketoapi"
 )
 
 type cliClient struct {
 	c *cmdx.CommandExecuter
 }
 
-func (g *cliClient) queryNamespaces(t require.TestingT) (res ketoapi.GetNamespacesResponse) {
-	if t, ok := t.(*testing.T); ok {
-		t.Skip("not implemented for the CLI")
-	}
+func (g *cliClient) queryNamespaces(t *testing.T) (res ketoapi.GetNamespacesResponse) {
+	t.Skip("not implemented for the CLI")
 	return
 }
 
 var _ client = (*cliClient)(nil)
 
-func (g *cliClient) oplCheckSyntax(t require.TestingT, _ []byte) []*ketoapi.ParseError {
-	if t, ok := t.(*testing.T); ok {
-		t.Skip("not implemented as a command yet")
-	}
+func (g *cliClient) oplCheckSyntax(t *testing.T, _ []byte) []*ketoapi.ParseError {
+	t.Skip("not implemented as a command yet")
 	return []*ketoapi.ParseError{}
 }
 
-func (g *cliClient) createTuple(t require.TestingT, r *ketoapi.RelationTuple) {
+func (g *cliClient) createTuple(t *testing.T, r *ketoapi.RelationTuple) {
 	tupleEnc, err := json.Marshal(r)
 	require.NoError(t, err)
 
@@ -88,7 +78,7 @@ func (g *cliClient) assembleQueryFlags(q *ketoapi.RelationQuery, opts []x.Pagina
 	return flags
 }
 
-func (g *cliClient) queryTuple(t require.TestingT, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter) *ketoapi.GetResponse {
+func (g *cliClient) queryTuple(t *testing.T, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter) *ketoapi.GetResponse {
 	out := g.c.ExecNoErr(t, append(g.assembleQueryFlags(q, opts), "relation-tuple", "get")...)
 
 	var resp ketoapi.GetResponse
@@ -97,13 +87,13 @@ func (g *cliClient) queryTuple(t require.TestingT, q *ketoapi.RelationQuery, opt
 	return &resp
 }
 
-func (g *cliClient) queryTupleErr(t require.TestingT, expected herodot.DefaultError, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter) {
+func (g *cliClient) queryTupleErr(t *testing.T, expected herodot.DefaultError, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter) {
 	stdErr := g.c.ExecExpectedErr(t, append(g.assembleQueryFlags(q, opts), "relation-tuple", "get")...)
 	assert.Contains(t, stdErr, expected.GRPCCodeField.String())
 	assert.Contains(t, stdErr, expected.Error())
 }
 
-func (g *cliClient) check(t require.TestingT, r *ketoapi.RelationTuple) bool {
+func (g *cliClient) check(t *testing.T, r *ketoapi.RelationTuple) bool {
 	var sub string
 	if r.SubjectID != nil {
 		sub = *r.SubjectID
@@ -116,27 +106,23 @@ func (g *cliClient) check(t require.TestingT, r *ketoapi.RelationTuple) bool {
 	return res.Allowed
 }
 
-func (g *cliClient) batchCheckErr(t require.TestingT, requestTuples []*ketoapi.RelationTuple,
-	expected herodot.DefaultError) {
-	if t, ok := t.(*testing.T); ok {
-		t.Skip("not implemented for the CLI")
-	}
+func (g *cliClient) batchCheckErr(t *testing.T, requestTuples []*ketoapi.RelationTuple, expected herodot.DefaultError) {
+	t.Skip("not implemented for the CLI")
 }
-func (g *cliClient) batchCheck(t require.TestingT, requestTuples []*ketoapi.RelationTuple) []checkResponse {
-	if t, ok := t.(*testing.T); ok {
-		t.Skip("not implemented for the CLI")
-	}
+
+func (g *cliClient) batchCheck(t *testing.T, requestTuples []*ketoapi.RelationTuple) []checkResponse {
+	t.Skip("not implemented for the CLI")
 	return nil
 }
 
-func (g *cliClient) expand(t require.TestingT, r *ketoapi.SubjectSet, depth int) *ketoapi.Tree[*ketoapi.RelationTuple] {
+func (g *cliClient) expand(t *testing.T, r *ketoapi.SubjectSet, depth int) *ketoapi.Tree[*ketoapi.RelationTuple] {
 	out := g.c.ExecNoErr(t, "expand", r.Relation, r.Namespace, r.Object, "--"+cliexpand.FlagMaxDepth, fmt.Sprintf("%d", depth), "--"+cmdx.FlagFormat, string(cmdx.FormatJSON))
 	res := ketoapi.Tree[*ketoapi.RelationTuple]{}
 	require.NoError(t, json.Unmarshal([]byte(out), &res))
 	return &res
 }
 
-func (g *cliClient) waitUntilLive(t require.TestingT) {
+func (g *cliClient) waitUntilLive(t *testing.T) {
 	flags := make([]string, len(g.c.PersistentArgs))
 	copy(flags, g.c.PersistentArgs)
 
@@ -154,7 +140,7 @@ func (g *cliClient) waitUntilLive(t require.TestingT) {
 	require.Equal(t, grpcHealthV1.HealthCheckResponse_SERVING.String()+"\n", out)
 }
 
-func (g *cliClient) deleteTuple(t require.TestingT, r *ketoapi.RelationTuple) {
+func (g *cliClient) deleteTuple(t *testing.T, r *ketoapi.RelationTuple) {
 	tupleEnc, err := json.Marshal(r)
 	require.NoError(t, err)
 
@@ -163,6 +149,6 @@ func (g *cliClient) deleteTuple(t require.TestingT, r *ketoapi.RelationTuple) {
 	assert.Len(t, stderr, 0, stdout)
 }
 
-func (g *cliClient) deleteAllTuples(t require.TestingT, q *ketoapi.RelationQuery) {
+func (g *cliClient) deleteAllTuples(t *testing.T, q *ketoapi.RelationQuery) {
 	_ = g.c.ExecNoErr(t, append(g.assembleQueryFlags(q, nil), "relation-tuple", "delete-all", "--force")...)
 }

internal/e2e/full_suit_test.go

@@ -28,21 +28,21 @@ import (
 type (
 	transactClient interface {
 		client
-		transactTuples(t require.TestingT, ins []*ketoapi.RelationTuple, del []*ketoapi.RelationTuple)
+		transactTuples(t *testing.T, ins []*ketoapi.RelationTuple, del []*ketoapi.RelationTuple)
 	}
 	client interface {
-		createTuple(t require.TestingT, r *ketoapi.RelationTuple)
-		deleteTuple(t require.TestingT, r *ketoapi.RelationTuple)
-		deleteAllTuples(t require.TestingT, q *ketoapi.RelationQuery)
-		queryTuple(t require.TestingT, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter) *ketoapi.GetResponse
-		queryTupleErr(t require.TestingT, expected herodot.DefaultError, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter)
-		check(t require.TestingT, r *ketoapi.RelationTuple) bool
-		batchCheck(t require.TestingT, r []*ketoapi.RelationTuple) []checkResponse
-		batchCheckErr(t require.TestingT, requestTuples []*ketoapi.RelationTuple, expected herodot.DefaultError)
-		expand(t require.TestingT, r *ketoapi.SubjectSet, depth int) *ketoapi.Tree[*ketoapi.RelationTuple]
-		oplCheckSyntax(t require.TestingT, content []byte) []*ketoapi.ParseError
-		waitUntilLive(t require.TestingT)
-		queryNamespaces(t require.TestingT) ketoapi.GetNamespacesResponse
+		createTuple(t *testing.T, r *ketoapi.RelationTuple)
+		deleteTuple(t *testing.T, r *ketoapi.RelationTuple)
+		deleteAllTuples(t *testing.T, q *ketoapi.RelationQuery)
+		queryTuple(t *testing.T, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter) *ketoapi.GetResponse
+		queryTupleErr(t *testing.T, expected herodot.DefaultError, q *ketoapi.RelationQuery, opts ...x.PaginationOptionSetter)
+		check(t *testing.T, r *ketoapi.RelationTuple) bool
+		batchCheck(t *testing.T, r []*ketoapi.RelationTuple) []checkResponse
+		batchCheckErr(t *testing.T, requestTuples []*ketoapi.RelationTuple, expected herodot.DefaultError)
+		expand(t *testing.T, r *ketoapi.SubjectSet, depth int) *ketoapi.Tree[*ketoapi.RelationTuple]
+		oplCheckSyntax(t *testing.T, content []byte) []*ketoapi.ParseError
+		waitUntilLive(t *testing.T)
+		queryNamespaces(t *testing.T) ketoapi.GetNamespacesResponse
 	}
 )
 
@@ -65,12 +65,11 @@ func Test(t *testing.T) {
 			// The test cases start here
 			// We execute every test with all clients available
 			for _, cl := range []client{
-				&grpcClient{
-					readRemote:      reg.Config(ctx).ReadAPIListenOn(),
-					writeRemote:     reg.Config(ctx).WriteAPIListenOn(),
-					oplSyntaxRemote: reg.Config(ctx).OPLSyntaxAPIListenOn(),
-					ctx:             ctx,
-				},
+				newGrpcClient(t, ctx,
+					reg.Config(ctx).ReadAPIListenOn(),
+					reg.Config(ctx).WriteAPIListenOn(),
+					reg.Config(ctx).OPLSyntaxAPIListenOn(),
+				),
 				&restClient{
 					readURL:      "http://" + reg.Config(ctx).ReadAPIListenOn(),
 					writeURL:     "http://" + reg.Config(ctx).WriteAPIListenOn(),
@@ -104,11 +103,11 @@ func Test(t *testing.T) {
 
 			t.Run("case=metrics are served", func(t *testing.T) {
 				t.Parallel()
-				(&grpcClient{
-					readRemote:  reg.Config(ctx).ReadAPIListenOn(),
-					writeRemote: reg.Config(ctx).WriteAPIListenOn(),
-					ctx:         ctx,
-				}).waitUntilLive(t)
+				newGrpcClient(t, ctx,
+					reg.Config(ctx).ReadAPIListenOn(),
+					reg.Config(ctx).WriteAPIListenOn(),
+					reg.Config(ctx).OPLSyntaxAPIListenOn(),
+				).waitUntilLive(t)
 
 				t.Run("case=on "+prometheus.MetricsPrometheusPath, func(t *testing.T) {
 					t.Parallel()
@@ -148,6 +147,7 @@ func TestServeConfig(t *testing.T) {
 		t.Log("Waiting for health check to be ready")
 		time.Sleep(10 * time.Millisecond)
 	}
+	t.Log("Health check is ready")
 
 	req, err := http.NewRequest(http.MethodOptions, "http://"+reg.Config(ctx).ReadAPIListenOn()+relationtuple.ReadRouteBase, nil)
 	require.NoError(t, err)