Add design proposal and implementation for ZedTokens table

cmd/createrole.go

@@ -6,7 +6,6 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"go.infratographer.com/x/crdbx"
-	"go.infratographer.com/x/events"
 	"go.infratographer.com/x/gidx"
 	"go.infratographer.com/x/viperx"
 
@@ -65,16 +64,6 @@ func createRole(ctx context.Context, cfg *config.AppConfig) {
 		logger.Fatalw("unable to initialize spicedb client", "error", err)
 	}
 
-	eventsConn, err := events.NewConnection(cfg.Events.Config, events.WithLogger(logger))
-	if err != nil {
-		logger.Fatalw("failed to initialize events", "error", err)
-	}
-
-	kv, err := initializeKV(cfg.Events, eventsConn)
-	if err != nil {
-		logger.Fatalw("failed to initialize KV", "error", err)
-	}
-
 	db, err := crdbx.NewDB(cfg.CRDB, cfg.Tracing.Enabled)
 	if err != nil {
 		logger.Fatalw("unable to initialize permissions-api database", "error", err)
@@ -109,17 +98,11 @@ func createRole(ctx context.Context, cfg *config.AppConfig) {
 		logger.Fatalw("error parsing subject ID", "error", err)
 	}
 
-	engine, err := query.NewEngine("infratographer", spiceClient, kv, store, query.WithPolicy(policy), query.WithLogger(logger))
+	engine, err := query.NewEngine("infratographer", spiceClient, store, query.WithPolicy(policy), query.WithLogger(logger))
 	if err != nil {
 		logger.Fatalw("error creating engine", "error", err)
 	}
 
-	defer func() {
-		if err := engine.Stop(); err != nil {
-			logger.Errorw("error stopping engine", "error", err)
-		}
-	}()
-
 	resource, err := engine.NewResourceFromID(resourceID)
 	if err != nil {
 		logger.Fatalw("error creating resource", "error", err)

cmd/server.go

@@ -8,7 +8,6 @@ import (
 	"go.infratographer.com/x/crdbx"
 	"go.infratographer.com/x/echojwtx"
 	"go.infratographer.com/x/echox"
-	"go.infratographer.com/x/events"
 	"go.infratographer.com/x/otelx"
 	"go.infratographer.com/x/versionx"
 	"go.uber.org/zap"
@@ -39,7 +38,6 @@ func init() {
 	echox.MustViperFlags(v, serverCmd.Flags(), apiDefaultListen)
 	otelx.MustViperFlags(v, serverCmd.Flags())
 	echojwtx.MustViperFlags(v, serverCmd.Flags())
-	events.MustViperFlags(v, serverCmd.Flags(), appName)
 }
 
 func serve(_ context.Context, cfg *config.AppConfig) {
@@ -53,16 +51,6 @@ func serve(_ context.Context, cfg *config.AppConfig) {
 		logger.Fatalw("unable to initialize spicedb client", "error", err)
 	}
 
-	eventsConn, err := events.NewConnection(cfg.Events.Config, events.WithLogger(logger))
-	if err != nil {
-		logger.Fatalw("failed to initialize events", "error", err)
-	}
-
-	kv, err := initializeKV(cfg.Events, eventsConn)
-	if err != nil {
-		logger.Fatalw("failed to initialize KV", "error", err)
-	}
-
 	db, err := crdbx.NewDB(cfg.CRDB, cfg.Tracing.Enabled)
 	if err != nil {
 		logger.Fatalw("unable to initialize permissions-api database", "error", err)
@@ -87,17 +75,11 @@ func serve(_ context.Context, cfg *config.AppConfig) {
 		logger.Fatalw("invalid spicedb policy", "error", err)
 	}
 
-	engine, err := query.NewEngine("infratographer", spiceClient, kv, store, query.WithPolicy(policy), query.WithLogger(logger))
+	engine, err := query.NewEngine("infratographer", spiceClient, store, query.WithPolicy(policy), query.WithLogger(logger))
 	if err != nil {
 		logger.Fatalw("error creating engine", "error", err)
 	}
 
-	defer func() {
-		if err := engine.Stop(); err != nil {
-			logger.Errorw("error stopping engine", "error", err)
-		}
-	}()
-
 	srv, err := echox.NewServer(
 		logger.Desugar(),
 		echox.ConfigFromViper(viper.GetViper()),

cmd/worker.go

@@ -59,11 +59,6 @@ func worker(ctx context.Context, cfg *config.AppConfig) {
 		logger.Fatalw("failed to initialize events", "error", err)
 	}
 
-	kv, err := initializeKV(cfg.Events, eventsConn)
-	if err != nil {
-		logger.Fatalw("failed to initialize KV", "error", err)
-	}
-
 	db, err := crdbx.NewDB(cfg.CRDB, cfg.Tracing.Enabled)
 	if err != nil {
 		logger.Fatalw("unable to initialize permissions-api database", "error", err)
@@ -88,7 +83,7 @@ func worker(ctx context.Context, cfg *config.AppConfig) {
 		logger.Fatalw("invalid spicedb policy", "error", err)
 	}
 
-	engine, err := query.NewEngine("infratographer", spiceClient, kv, store, query.WithPolicy(policy))
+	engine, err := query.NewEngine("infratographer", spiceClient, store, query.WithPolicy(policy))
 	if err != nil {
 		logger.Fatalw("error creating engine", "error", err)
 	}

docs/20240515-zedtokens.md

@@ -0,0 +1,46 @@
+# Consistency with ZedTokens
+
+## Overview
+
+In SpiceDB, resources are arranged in a graph where different services may be contributing to that graph. For example, permissions-api itself manages roles and role bindings, while tenant-api manages tenants. Resources are updated over NATS request/reply.
+
+SpiceDB breaks updates down into quantized revisions of configurable duration, meaning ACL updates within a given revision window are not visible by default until the next revision begins. Using ZedTokens, however, clients can request data at least as fresh as a given exact point in time. This allows for clients to control the consistency they need for a given SpiceDB request on a per-request basis.
+
+In general, services expect that immediately after updating relationships in permissions-api, they can make authorization decisions based on data at least as fresh as when they made those updates.
+
+## Goals
+
+This proposal is meant to achieve the following goals:
+
+* Permissions checks made on a resource after it is updated must use data at least as fresh as the last update to that resource
+
+## Non-goals
+
+This proposal is not meant to achieve the following goals:
+
+* Updates to the graph are globally and immediately propagated
+
+## Proposed solution
+
+To mitigate this issue, permissions-api can be updated to use a table in CRDB to populate ZedTokens for recently updated resources. By doing this, permissions-api becomes responsible for determining the freshness of lookups. On permissions checks for a given resource, the following consistency strategy is used:
+
+* If a ZedToken exists for that resource, use `at_least_as_fresh` with the given ZedToken
+* If no ZedToken exists, use `minimize_latency`
+
+The advantage of this approach is that it keeps management of consistency within permissions-api, meaning future changes do not (necessarily) require updates to other services, nor does it change current API semantics. Additionally, it does not introduce new dependencies to permissions-api, instead leveraging CRDB's availability and fault tolerance guarantees.
+
+## Constraints and limitations
+
+As designed, this only provides immediately visible updates for a given resource; other indirectly affected resources are not updated (but could be in the future). Additionally, given CRDB leaseholder behavior, multi-region deployments may see longer durations when performing permissions checks if the ZedToken table is not a global table.
+
+## Alternatives considered
+
+### Stream ZedToken updates using the SpiceDB Watch API
+
+We could minimize lookup time for ZedTokens by using the SpiceDB Watch API and streaming updates to all permissions-api replicas. However, this means updates will not be immediately visible for a given resource globally. While tokens could be stored in clients (i.e., using the IAM runtime), this merely pushes the complexity of managing ZedTokens to downstream services.
+
+### Store ZedTokens using NATS KV
+
+ZedTokens could be stored using KV in NATS as a coordination mechanism between servers globally. NATS clusters are in general meant to have very low latency between servers, where a single cluster coordinates a given stream (and thus a given KV bucket). This means that for immediate updates to resources using KV, all permissions-api replicas either need to read from a single NATS cluster (introducing latency as a function of distance from the cluster) or use a large cluster that spans many regions. The latter approach is [explicitly discouraged][nats-discussion] by NATS maintainers.
+
+[nats-discussion]: https://github.com/nats-io/nats-server/discussions/5317#discussioncomment-9138192

go.mod

@@ -7,10 +7,8 @@ require (
 	github.com/authzed/grpcutil v0.0.0-20240123194739-2ea1e3d2d98b
 	github.com/cockroachdb/cockroach-go/v2 v2.3.7
 	github.com/go-jose/go-jose/v4 v4.0.1
-	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/jackc/pgx/v5 v5.5.5
 	github.com/labstack/echo/v4 v4.11.4
-	github.com/nats-io/nats.go v1.34.1
 	github.com/pkg/errors v0.9.1
 	github.com/pressly/goose/v3 v3.19.2
 	github.com/spf13/cobra v1.8.0
@@ -68,6 +66,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/nats-io/jwt/v2 v2.5.5 // indirect
 	github.com/nats-io/nats-server/v2 v2.10.12 // indirect
+	github.com/nats-io/nats.go v1.34.1 // indirect
 	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.0 // indirect

go.sum

@@ -126,6 +126,7 @@ github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 h1:UH//fgunKIs4JdUbpDl1VZCDa
 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0/go.mod h1:g5qyo/la0ALbONm6Vbp88Yd8NsDy6rZz+RcrMPxvld8=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 h1:/c3QmbOGMGTOumP2iT/rCwB7b0QDGLKzqOmktBjT+Is=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1/go.mod h1:5SN9VR2LTsRFsrEC6FHgRbTWrTHu6tqPeKxEQv15giM=
+github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=

internal/query/relations.go

@@ -113,7 +113,7 @@ func (e *engine) SubjectHasPermission(ctx context.Context, subject types.Resourc
 
 	defer span.End()
 
-	consistency, consName := e.determineConsistency(resource)
+	consistency, consName := e.determineConsistency(ctx, resource)
 	span.SetAttributes(
 		attribute.String(
 			"permissions.consistency",
@@ -284,7 +284,7 @@ func (e *engine) CreateRelationships(ctx context.Context, rels []types.Relations
 		}
 	}
 
-	relUpdates := e.relationshipsToUpdates(rels)
+	relUpdates := e.relationshipsToUpdates(rels, pb.RelationshipUpdate_OPERATION_TOUCH)
 
 	request := &pb.WriteRelationshipsRequest{
 		Updates: relUpdates,
@@ -591,15 +591,15 @@ func (e *engine) roleResourceRelationshipsTouchDelete(roleResource, resource typ
 	return rels
 }
 
-func (e *engine) relationshipsToUpdates(rels []types.Relationship) []*pb.RelationshipUpdate {
+func (e *engine) relationshipsToUpdates(rels []types.Relationship, operation pb.RelationshipUpdate_Operation) []*pb.RelationshipUpdate {
 	relUpdates := make([]*pb.RelationshipUpdate, len(rels))
 
 	for i, rel := range rels {
 		subjRef := resourceToSpiceDBRef(e.namespace, rel.Subject)
 		resRef := resourceToSpiceDBRef(e.namespace, rel.Resource)
 
 		relUpdates[i] = &pb.RelationshipUpdate{
-			Operation: pb.RelationshipUpdate_OPERATION_TOUCH,
+			Operation: operation,
 			Relationship: &pb.Relationship{
 				Resource: resRef,
 				Relation: rel.Relation,
@@ -677,65 +677,22 @@ func (e *engine) DeleteRelationships(ctx context.Context, relationships ...types
 		return multierr.Combine(errors...)
 	}
 
-	errors = []error{}
-
-	var (
-		complete []types.Relationship
-		dErr     error
-		cErr     error
-	)
-
-	span.AddEvent("deleting relationships")
-
-	for i, relationship := range relationships {
-		resType := e.namespace + "/" + relationship.Resource.Type
-		subjType := e.namespace + "/" + relationship.Subject.Type
-
-		filter := &pb.RelationshipFilter{
-			ResourceType:       resType,
-			OptionalResourceId: relationship.Resource.ID.String(),
-			OptionalRelation:   relationship.Relation,
-			OptionalSubjectFilter: &pb.SubjectFilter{
-				SubjectType:       subjType,
-				OptionalSubjectId: relationship.Subject.ID.String(),
-			},
-		}
-
-		if dErr = e.deleteRelationships(ctx, filter); dErr != nil {
-			e.logger.Errorf("%w: failed to delete relationship %d reverting %d completed deletes", dErr, i, len(complete))
+	relUpdates := e.relationshipsToUpdates(relationships, pb.RelationshipUpdate_OPERATION_DELETE)
 
-			err := fmt.Errorf("%w: failed to delete relationship %d", dErr, i)
-
-			span.RecordError(err)
-
-			errors = append(errors, err)
-
-			break
-		}
-
-		complete = append(complete, relationship)
+	request := &pb.WriteRelationshipsRequest{
+		Updates: relUpdates,
 	}
 
-	if len(errors) != 0 {
-		span.SetStatus(codes.Error, "error occurred deleting relationships")
-
-		if len(complete) != 0 {
-			span.AddEvent("recreating deleted relationships")
-
-			if cErr = e.CreateRelationships(ctx, complete); cErr != nil {
-				e.logger.Error("%w: failed to revert %d deleted relationships", cErr, len(complete))
-
-				err := fmt.Errorf("%w: failed to revert deleted relationships", cErr)
-
-				span.RecordError(err)
-
-				errors = append(errors, err)
-			}
-		}
+	resp, err := e.client.WriteRelationships(ctx, request)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
 
-		return multierr.Combine(errors...)
+		return err
 	}
 
+	e.updateRelationshipZedTokens(ctx, relationships, resp.WrittenAt.Token)
+
 	return nil
 }
 
@@ -1266,7 +1223,12 @@ func (e *engine) rollbackUpdates(ctx context.Context, updates []*pb.Relationship
 		})
 	}
 
-	return e.applyUpdates(ctx, rollbacks)
+	_, err := e.client.WriteRelationships(ctx, &pb.WriteRelationshipsRequest{Updates: rollbacks})
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // applyUpdates is a wrapper function around the spiceDB WriteRelationships method

internal/query/relations_test.go

@@ -6,12 +6,10 @@ import (
 
 	pb "github.com/authzed/authzed-go/proto/authzed/api/v1"
 	"github.com/authzed/authzed-go/v1"
-	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"go.infratographer.com/x/gidx"
-	"go.infratographer.com/x/testing/eventtools"
 
 	"go.infratographer.com/permissions-api/internal/iapl"
 	"go.infratographer.com/permissions-api/internal/spicedbx"
@@ -40,30 +38,16 @@ func testEngine(ctx context.Context, t *testing.T, namespace string, policy iapl
 	_, err = client.WriteSchema(ctx, request)
 	require.NoError(t, err)
 
-	natsSrv, err := eventtools.NewNatsServer()
-	require.NoError(t, err)
-
-	kvCfg := nats.KeyValueConfig{
-		Bucket: "zedtokens",
-	}
-
-	kv, err := natsSrv.JetStream.CreateKeyValue(&kvCfg)
-	require.NoError(t, err)
-
 	t.Cleanup(func() {
 		cleanDB(ctx, t, client, namespace, policy)
 		cleanStore()
 	})
 
 	// We call the constructor here to ensure the engine is created appropriately, but
 	// then return the underlying type so we can do testing with it.
-	out, err := NewEngine(namespace, client, kv, store, WithPolicy(policy))
+	out, err := NewEngine(namespace, client, store, WithPolicy(policy))
 	require.NoError(t, err)
 
-	t.Cleanup(func() {
-		out.Stop() //nolint:errcheck
-	})
-
 	return out.(*engine)
 }