-
Notifications
You must be signed in to change notification settings - Fork 290
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- refactors memdb caveat test and turns it into a datastore test. Datastores not implementing the interface are skipped - adds migration to add caveat table for PSQL - makes psql TX implements CaveatStorer, following same strategy for MVCC - adds context to ReadCaveatByName to follow same signature as ReadNamespace - adjusts memdb to avoid upserts on duplicate caveats, and align with PSQL - implement missing DeleteCaveat test, and implement that logic for memdb and postgres
- Loading branch information
1 parent
aa3712f
commit cb7b956
Showing
21 changed files
with
432 additions
and
65 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| package postgres | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
|
|
||
| "github.com/authzed/spicedb/internal/datastore/common" | ||
| "github.com/authzed/spicedb/pkg/datastore" | ||
| core "github.com/authzed/spicedb/pkg/proto/core/v1" | ||
|
|
||
| sq "github.com/Masterminds/squirrel" | ||
| "github.com/jackc/pgx/v4" | ||
| "go.opentelemetry.io/otel/attribute" | ||
| "go.opentelemetry.io/otel/trace" | ||
| ) | ||
|
|
||
| var ( | ||
| writeCaveat = psql.Insert(tableCaveat).Columns(colCaveatName, colCaveatExpression) | ||
| writeCaveatDeprecated = psql.Insert(tableCaveat).Columns(colCaveatName, colCaveatExpression, colCreatedTxnDeprecated) | ||
| readCaveat = psql.Select(colCaveatExpression).From(tableCaveat) | ||
| deleteCaveat = psql.Update(tableCaveat).Where(sq.Eq{colDeletedXid: liveDeletedTxnID}) | ||
| deleteCaveatDeprecated = psql.Update(tableCaveat).Where(sq.Eq{colDeletedTxnDeprecated: liveDeletedTxnID}) | ||
| ) | ||
|
|
||
| func (r *pgReader) ReadCaveatByName(ctx context.Context, name string) (*core.Caveat, error) { | ||
| ctx, span := tracer.Start(ctx, "ReadCaveatByName", trace.WithAttributes(attribute.String("name", name))) | ||
| defer span.End() | ||
|
|
||
| filteredReadCaveat := r.filterer(readCaveat) | ||
| sql, args, err := filteredReadCaveat.Where(sq.Eq{colCaveatName: name}).ToSql() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| tx, txCleanup, err := r.txSource(ctx) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("unable to read caveat: %w", err) | ||
| } | ||
| defer txCleanup(ctx) | ||
|
|
||
| var expr []byte | ||
| err = tx.QueryRow(ctx, sql, args...).Scan(&expr) | ||
| if err != nil { | ||
| if errors.Is(err, pgx.ErrNoRows) { | ||
| return nil, datastore.NewCaveatNameNotFoundErr(name) | ||
| } | ||
| return nil, err | ||
| } | ||
| return &core.Caveat{ | ||
| Name: name, | ||
| Expression: expr, | ||
| }, nil | ||
| } | ||
|
|
||
| func (rwt *pgReadWriteTXN) WriteCaveats(caveats []*core.Caveat) error { | ||
| ctx, span := tracer.Start(datastore.SeparateContextWithTracing(rwt.ctx), "WriteCaveats") | ||
| defer span.End() | ||
|
|
||
| deletedCaveatClause := sq.Or{} | ||
| write := writeCaveat | ||
| // TODO remove once the ID->XID migrations are all complete | ||
| if rwt.migrationPhase == writeBothReadNew || rwt.migrationPhase == writeBothReadOld { | ||
| write = writeCaveatDeprecated | ||
| } | ||
| writtenCaveatNames := make([]string, 0, len(caveats)) | ||
| for _, caveat := range caveats { | ||
| deletedCaveatClause = append(deletedCaveatClause, sq.Eq{colCaveatName: caveat.Name}) | ||
| valuesToWrite := []any{caveat.Name, caveat.Expression} | ||
| // TODO remove once the ID->XID migrations are all complete | ||
| if rwt.migrationPhase == writeBothReadNew || rwt.migrationPhase == writeBothReadOld { | ||
| valuesToWrite = append(valuesToWrite, rwt.newXID.Uint) | ||
| } | ||
| write = write.Values(valuesToWrite...) | ||
| writtenCaveatNames = append(writtenCaveatNames, caveat.Name) | ||
| } | ||
| span.SetAttributes(common.CaveatNameKey.StringSlice(writtenCaveatNames)) | ||
|
|
||
| // mark current caveats as deleted | ||
| err := rwt.deleteCaveatsWithClause(ctx, deletedCaveatClause) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // store the new caveat revision | ||
| sql, args, err := write.ToSql() | ||
| if err != nil { | ||
| return fmt.Errorf("unable to write new caveat revision: %w", err) | ||
| } | ||
| if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { | ||
| return fmt.Errorf("unable to write new caveat revision: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (rwt *pgReadWriteTXN) DeleteCaveats(caveats []*core.Caveat) error { | ||
| ctx, span := tracer.Start(datastore.SeparateContextWithTracing(rwt.ctx), "DeleteCaveats") | ||
| defer span.End() | ||
|
|
||
| deletedCaveatClause := sq.Or{} | ||
| deletedCaveatNames := make([]string, 0, len(caveats)) | ||
| for _, caveat := range caveats { | ||
| deletedCaveatClause = append(deletedCaveatClause, sq.Eq{colCaveatName: caveat.Name}) | ||
| deletedCaveatNames = append(deletedCaveatNames, caveat.Name) | ||
| } | ||
| span.SetAttributes(common.CaveatNameKey.StringSlice(deletedCaveatNames)) | ||
|
|
||
| // mark current caveats as deleted | ||
| return rwt.deleteCaveatsWithClause(ctx, deletedCaveatClause) | ||
| } | ||
|
|
||
| func (rwt *pgReadWriteTXN) deleteCaveatsWithClause(ctx context.Context, deleteClauses sq.Or) error { | ||
| sql, args, err := deleteCaveat. | ||
| Set(colDeletedXid, rwt.newXID). | ||
| Where(sq.And{sq.Eq{colDeletedXid: liveDeletedTxnID}, deleteClauses}). | ||
| ToSql() | ||
| if err != nil { | ||
| return fmt.Errorf("unable to mark previous caveat revisions as deleted: %w", err) | ||
| } | ||
|
|
||
| // TODO remove once the ID->XID migrations are all complete | ||
| if rwt.migrationPhase == writeBothReadNew || rwt.migrationPhase == writeBothReadOld { | ||
| baseQuery := deleteCaveat | ||
| if rwt.migrationPhase == writeBothReadOld { | ||
| baseQuery = deleteCaveatDeprecated | ||
| } | ||
|
|
||
| sql, args, err = baseQuery. | ||
| Where(deleteClauses). | ||
| Set(colDeletedTxnDeprecated, rwt.newXID.Uint). | ||
| Set(colDeletedXid, rwt.newXID). | ||
| ToSql() | ||
| if err != nil { | ||
| return fmt.Errorf("unable to mark previous caveat revisions as deleted: %w", err) | ||
| } | ||
| } | ||
|
|
||
| if _, err := rwt.tx.Exec(ctx, sql, args...); err != nil { | ||
| return fmt.Errorf("unable to mark previous caveat revisions as deleted: %w", err) | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
36 changes: 36 additions & 0 deletions
36
internal/datastore/postgres/migrations/zz_migration.0009_caveat.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package migrations | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/jackc/pgx/v4" | ||
| ) | ||
|
|
||
| var caveatStatements = []string{ | ||
| `CREATE TABLE caveat ( | ||
| name VARCHAR NOT NULL, | ||
| expression BYTEA NOT NULL, | ||
| created_transaction BIGINT NOT NULL, | ||
| deleted_transaction BIGINT NOT NULL DEFAULT '9223372036854775807', | ||
| CONSTRAINT pk_caveat_v1 PRIMARY KEY (name, deleted_transaction), | ||
| CONSTRAINT uq_caveat_v1 UNIQUE (name, created_transaction, deleted_transaction));`, | ||
| `ALTER TABLE relation_tuple | ||
| ADD COLUMN caveat_name VARCHAR, | ||
| ADD COLUMN caveat_context JSONB;`, | ||
| } | ||
|
|
||
| func init() { | ||
| if err := DatabaseMigrations.Register("add-caveats", "add-ns-config-id", | ||
| noNonatomicMigration, | ||
| func(ctx context.Context, tx pgx.Tx) error { | ||
| for _, stmt := range caveatStatements { | ||
| if _, err := tx.Exec(ctx, stmt); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| }); err != nil { | ||
| panic("failed to register migration: " + err.Error()) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.