10924732005

Committed 18 Sep 2024 02:54PM UTC coverage: 79.97% (+0.3%) from 79.691%

Build # 10924732005

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1603 from Permify/lookup-improvements

feat: add scope of entityIds to lookup-entity api filter

Run Details

379 of 478 new or added lines in 13 files covered. (79.29%)

31 existing lines in 4 files now uncovered.

7993 of 9995 relevant lines covered (79.97%)

119.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

37.84

/internal/storage/postgres/utils/common.go

package utils

import (
        "context"
        "errors"
        "fmt"
        "log/slog"
        "math"
        "strings"
        "time"

        "go.opentelemetry.io/otel/codes"
        "golang.org/x/exp/rand"

        "go.opentelemetry.io/otel/trace"

        "github.com/Masterminds/squirrel"

        base "github.com/Permify/permify/pkg/pb/base/v1"
)

const (
        TransactionTemplate       = `INSERT INTO transactions (tenant_id) VALUES ($1) RETURNING id`
        InsertTenantTemplate      = `INSERT INTO tenants (id, name) VALUES ($1, $2) RETURNING created_at`
        DeleteTenantTemplate      = `DELETE FROM tenants WHERE id = $1 RETURNING name, created_at`
        DeleteAllByTenantTemplate = `DELETE FROM %s WHERE tenant_id = $1`
)

// SnapshotQuery adds conditions to a SELECT query for checking transaction visibility based on created and expired transaction IDs.
// The query checks if transactions are visible in a snapshot associated with the provided value.
func SnapshotQuery(sl squirrel.SelectBuilder, value uint64) squirrel.SelectBuilder {
        // Convert the value to a string once to reduce redundant calls to fmt.Sprintf.
        valStr := fmt.Sprintf("'%v'::xid8", value)

        // Create a subquery for the snapshot associated with the provided value.
        snapshotQuery := fmt.Sprintf("(select snapshot from transactions where id = %s)", valStr)

        // Create an expression to check if a transaction with a specific created_tx_id is visible in the snapshot.
        visibilityExpr := squirrel.Expr(fmt.Sprintf("pg_visible_in_snapshot(created_tx_id, %s) = true", snapshotQuery))
        // Create an expression to check if the created_tx_id is equal to the provided value.
        createdExpr := squirrel.Expr(fmt.Sprintf("created_tx_id = %s", valStr))
        // Use OR condition for the created expressions.
        createdWhere := squirrel.Or{visibilityExpr, createdExpr}

        // Create an expression to check if a transaction with a specific expired_tx_id is not visible in the snapshot.
        expiredVisibilityExpr := squirrel.Expr(fmt.Sprintf("pg_visible_in_snapshot(expired_tx_id, %s) = false", snapshotQuery))
        // Create an expression to check if the expired_tx_id is equal to zero.
        expiredZeroExpr := squirrel.Expr("expired_tx_id = '0'::xid8")
        // Create an expression to check if the expired_tx_id is not equal to the provided value.
        expiredNotExpr := squirrel.Expr(fmt.Sprintf("expired_tx_id <> %s", valStr))
        // Use AND condition for the expired expressions, checking both visibility and non-equality with value.
        expiredWhere := squirrel.And{squirrel.Or{expiredVisibilityExpr, expiredZeroExpr}, expiredNotExpr}

        // Add the created and expired conditions to the SELECT query.
        return sl.Where(createdWhere).Where(expiredWhere)
}

// snapshotQuery function generates two strings representing conditions to be applied in a SQL query to filter data based on visibility of transactions.
func snapshotQuery(value uint64) (string, string) {
        // Convert the provided value into a string format suitable for our SQL query, formatted as a transaction ID.
        valStr := fmt.Sprintf("'%v'::xid8", value)

        // Create a subquery that fetches the snapshot associated with the transaction ID.
        snapshotQ := fmt.Sprintf("(SELECT snapshot FROM transactions WHERE id = %s)", valStr)

        // Create an expression that checks whether a transaction (represented by 'created_tx_id') is visible in the snapshot.
        visibilityExpr := fmt.Sprintf("pg_visible_in_snapshot(created_tx_id, %s) = true", snapshotQ)
        // Create an expression that checks if the 'created_tx_id' is the same as our transaction ID.
        createdExpr := fmt.Sprintf("created_tx_id = %s", valStr)
        // Combine these expressions to form a condition. A row will satisfy this condition if either of the expressions are true.
        createdWhere := fmt.Sprintf("(%s OR %s)", visibilityExpr, createdExpr)

        // Create an expression that checks whether a transaction (represented by 'expired_tx_id') is not visible in the snapshot.
        expiredVisibilityExpr := fmt.Sprintf("pg_visible_in_snapshot(expired_tx_id, %s) = false", snapshotQ)
        // Create an expression that checks if the 'expired_tx_id' is zero. This handles cases where the transaction hasn't expired.
        expiredZeroExpr := "expired_tx_id = '0'::xid8"
        // Create an expression that checks if the 'expired_tx_id' is not the same as our transaction ID.
        expiredNotExpr := fmt.Sprintf("expired_tx_id <> %s", valStr)
        // Combine these expressions to form a condition. A row will satisfy this condition if the first set of expressions are true (either the transaction hasn't expired, or if it has, it's not visible in the snapshot) and the second expression is also true (the 'expired_tx_id' is not the same as our transaction ID).
        expiredWhere := fmt.Sprintf("(%s AND %s)", fmt.Sprintf("(%s OR %s)", expiredVisibilityExpr, expiredZeroExpr), expiredNotExpr)

        // Return the conditions for both 'created' and 'expired' transactions. These can be used in a WHERE clause of a SQL query to filter results.
        return createdWhere, expiredWhere
}

// GenerateGCQuery generates a Squirrel DELETE query builder for garbage collection.
// It constructs a query to delete expired records from the specified table
// based on the provided value, which represents a transaction ID.
func GenerateGCQuery(table string, value uint64) squirrel.DeleteBuilder {
        // Convert the provided value into a string format suitable for our SQL query, formatted as a transaction ID.
        valStr := fmt.Sprintf("'%v'::xid8", value)

        // Create a Squirrel DELETE builder for the specified table.
        deleteBuilder := squirrel.Delete(table)

        // Create an expression to check if 'expired_tx_id' is not equal to '0' (not expired).
        expiredZeroExpr := squirrel.Expr("expired_tx_id <> '0'::xid8")

        // Create an expression to check if 'expired_tx_id' is less than the provided value (before the cutoff).
        beforeExpr := squirrel.Expr(fmt.Sprintf("expired_tx_id < %s", valStr))

        // Add the WHERE clauses to the DELETE query builder to filter and delete expired data.
        return deleteBuilder.Where(expiredZeroExpr).Where(beforeExpr)
}

// HandleError records an error in the given span, logs the error, and returns a standardized error.
// This function is used for consistent error handling across different parts of the application.
func HandleError(ctx context.Context, span trace.Span, err error, errorCode base.ErrorCode) error {
        // Check if the error is context-related
        if IsContextRelatedError(ctx, err) {
                slog.DebugContext(ctx, "A context-related error occurred",
                        slog.String("error", err.Error()))
                return errors.New(base.ErrorCode_ERROR_CODE_CANCELLED.String())
        }

        // Check if the error is serialization-related
        if IsSerializationRelatedError(err) {
                slog.DebugContext(ctx, "A serialization-related error occurred",
                        slog.String("error", err.Error()))
                return errors.New(base.ErrorCode_ERROR_CODE_SERIALIZATION.String())
        }

        // For all other types of errors, log them at the error level and record them in the span
        slog.ErrorContext(ctx, "An operational error occurred",
                slog.Any("error", err))
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())

        // Return a new error with the standard error code provided
        return errors.New(errorCode.String())
}

// IsContextRelatedError checks if the error is due to context cancellation, deadline exceedance, or closed connection
func IsContextRelatedError(ctx context.Context, err error) bool {
        if errors.Is(ctx.Err(), context.Canceled) || errors.Is(ctx.Err(), context.DeadlineExceeded) {
                return true
        }
        if errors.Is(err, context.Canceled) ||
                errors.Is(err, context.DeadlineExceeded) ||
                strings.Contains(err.Error(), "conn closed") {
                return true
        }
        return false
}

// IsSerializationRelatedError checks if the error is a serialization failure, typically in database transactions.
func IsSerializationRelatedError(err error) bool {
        if strings.Contains(err.Error(), "could not serialize") ||
                strings.Contains(err.Error(), "duplicate key value") {
                return true
        }
        return false
}

// WaitWithBackoff implements an exponential backoff strategy with jitter for retries.
// It waits for a calculated duration or until the context is cancelled, whichever comes first.
func WaitWithBackoff(ctx context.Context, tenantID string, retries int) {
        backoff := time.Duration(math.Min(float64(20*time.Millisecond)*math.Pow(2, float64(retries)), float64(1*time.Second)))
        jitter := time.Duration(rand.Float64() * float64(backoff) * 0.5)
        nextBackoff := backoff + jitter
        slog.WarnContext(ctx, "waiting before retry", slog.String("tenant_id", tenantID), slog.Int64("backoff_duration", nextBackoff.Milliseconds()))
        select {
        case <-time.After(nextBackoff):
        case <-ctx.Done():
        }
}

1	package utils
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7	"log/slog"
8	"math"
9	"strings"
10	"time"
11
12	"go.opentelemetry.io/otel/codes"
13	"golang.org/x/exp/rand"
14
15	"go.opentelemetry.io/otel/trace"
16
17	"github.com/Masterminds/squirrel"
18
19	base "github.com/Permify/permify/pkg/pb/base/v1"
20	)
21
22	const (
23	TransactionTemplate = `INSERT INTO transactions (tenant_id) VALUES ($1) RETURNING id`
24	InsertTenantTemplate = `INSERT INTO tenants (id, name) VALUES ($1, $2) RETURNING created_at`
25	DeleteTenantTemplate = `DELETE FROM tenants WHERE id = $1 RETURNING name, created_at`
26	DeleteAllByTenantTemplate = `DELETE FROM %s WHERE tenant_id = $1`
27	)
28
29	// SnapshotQuery adds conditions to a SELECT query for checking transaction visibility based on created and expired transaction IDs.
30	// The query checks if transactions are visible in a snapshot associated with the provided value.
31	func SnapshotQuery(sl squirrel.SelectBuilder, value uint64) squirrel.SelectBuilder {	1✔
32	// Convert the value to a string once to reduce redundant calls to fmt.Sprintf.	1✔
33	valStr := fmt.Sprintf("'%v'::xid8", value)	1✔
34		1✔
35	// Create a subquery for the snapshot associated with the provided value.	1✔
36	snapshotQuery := fmt.Sprintf("(select snapshot from transactions where id = %s)", valStr)	1✔
37		1✔
38	// Create an expression to check if a transaction with a specific created_tx_id is visible in the snapshot.	1✔
39	visibilityExpr := squirrel.Expr(fmt.Sprintf("pg_visible_in_snapshot(created_tx_id, %s) = true", snapshotQuery))	1✔
40	// Create an expression to check if the created_tx_id is equal to the provided value.	1✔
41	createdExpr := squirrel.Expr(fmt.Sprintf("created_tx_id = %s", valStr))	1✔
42	// Use OR condition for the created expressions.	1✔
43	createdWhere := squirrel.Or{visibilityExpr, createdExpr}	1✔
44		1✔
45	// Create an expression to check if a transaction with a specific expired_tx_id is not visible in the snapshot.	1✔
46	expiredVisibilityExpr := squirrel.Expr(fmt.Sprintf("pg_visible_in_snapshot(expired_tx_id, %s) = false", snapshotQuery))	1✔
47	// Create an expression to check if the expired_tx_id is equal to zero.	1✔
48	expiredZeroExpr := squirrel.Expr("expired_tx_id = '0'::xid8")	1✔
49	// Create an expression to check if the expired_tx_id is not equal to the provided value.	1✔
50	expiredNotExpr := squirrel.Expr(fmt.Sprintf("expired_tx_id <> %s", valStr))	1✔
51	// Use AND condition for the expired expressions, checking both visibility and non-equality with value.	1✔
52	expiredWhere := squirrel.And{squirrel.Or{expiredVisibilityExpr, expiredZeroExpr}, expiredNotExpr}	1✔
53		1✔
54	// Add the created and expired conditions to the SELECT query.	1✔
55	return sl.Where(createdWhere).Where(expiredWhere)	1✔
56	}	1✔
57
58	// snapshotQuery function generates two strings representing conditions to be applied in a SQL query to filter data based on visibility of transactions.
UNCOV 59	func snapshotQuery(value uint64) (string, string) {	×
UNCOV 60	// Convert the provided value into a string format suitable for our SQL query, formatted as a transaction ID.	×
UNCOV 61	valStr := fmt.Sprintf("'%v'::xid8", value)	×
UNCOV 62		×
UNCOV 63	// Create a subquery that fetches the snapshot associated with the transaction ID.	×
UNCOV 64	snapshotQ := fmt.Sprintf("(SELECT snapshot FROM transactions WHERE id = %s)", valStr)	×
UNCOV 65		×
UNCOV 66	// Create an expression that checks whether a transaction (represented by 'created_tx_id') is visible in the snapshot.	×
UNCOV 67	visibilityExpr := fmt.Sprintf("pg_visible_in_snapshot(created_tx_id, %s) = true", snapshotQ)	×
UNCOV 68	// Create an expression that checks if the 'created_tx_id' is the same as our transaction ID.	×
UNCOV 69	createdExpr := fmt.Sprintf("created_tx_id = %s", valStr)	×
UNCOV 70	// Combine these expressions to form a condition. A row will satisfy this condition if either of the expressions are true.	×
UNCOV 71	createdWhere := fmt.Sprintf("(%s OR %s)", visibilityExpr, createdExpr)	×
UNCOV 72		×
UNCOV 73	// Create an expression that checks whether a transaction (represented by 'expired_tx_id') is not visible in the snapshot.	×
UNCOV 74	expiredVisibilityExpr := fmt.Sprintf("pg_visible_in_snapshot(expired_tx_id, %s) = false", snapshotQ)	×
UNCOV 75	// Create an expression that checks if the 'expired_tx_id' is zero. This handles cases where the transaction hasn't expired.	×
UNCOV 76	expiredZeroExpr := "expired_tx_id = '0'::xid8"	×
UNCOV 77	// Create an expression that checks if the 'expired_tx_id' is not the same as our transaction ID.	×
UNCOV 78	expiredNotExpr := fmt.Sprintf("expired_tx_id <> %s", valStr)	×
UNCOV 79	// Combine these expressions to form a condition. A row will satisfy this condition if the first set of expressions are true (either the transaction hasn't expired, or if it has, it's not visible in the snapshot) and the second expression is also true (the 'expired_tx_id' is not the same as our transaction ID).	×
UNCOV 80	expiredWhere := fmt.Sprintf("(%s AND %s)", fmt.Sprintf("(%s OR %s)", expiredVisibilityExpr, expiredZeroExpr), expiredNotExpr)	×
UNCOV 81		×
UNCOV 82	// Return the conditions for both 'created' and 'expired' transactions. These can be used in a WHERE clause of a SQL query to filter results.	×
UNCOV 83	return createdWhere, expiredWhere	×
UNCOV 84	}	×
85
86	// GenerateGCQuery generates a Squirrel DELETE query builder for garbage collection.
87	// It constructs a query to delete expired records from the specified table
88	// based on the provided value, which represents a transaction ID.
89	func GenerateGCQuery(table string, value uint64) squirrel.DeleteBuilder {	1✔
90	// Convert the provided value into a string format suitable for our SQL query, formatted as a transaction ID.	1✔
91	valStr := fmt.Sprintf("'%v'::xid8", value)	1✔
92		1✔
93	// Create a Squirrel DELETE builder for the specified table.	1✔
94	deleteBuilder := squirrel.Delete(table)	1✔
95		1✔
96	// Create an expression to check if 'expired_tx_id' is not equal to '0' (not expired).	1✔
97	expiredZeroExpr := squirrel.Expr("expired_tx_id <> '0'::xid8")	1✔
98		1✔
99	// Create an expression to check if 'expired_tx_id' is less than the provided value (before the cutoff).	1✔
100	beforeExpr := squirrel.Expr(fmt.Sprintf("expired_tx_id < %s", valStr))	1✔
101		1✔
102	// Add the WHERE clauses to the DELETE query builder to filter and delete expired data.	1✔
103	return deleteBuilder.Where(expiredZeroExpr).Where(beforeExpr)	1✔
104	}	1✔
105
106	// HandleError records an error in the given span, logs the error, and returns a standardized error.
107	// This function is used for consistent error handling across different parts of the application.
108	func HandleError(ctx context.Context, span trace.Span, err error, errorCode base.ErrorCode) error {	×
109	// Check if the error is context-related	×
110	if IsContextRelatedError(ctx, err) {	×
111	slog.DebugContext(ctx, "A context-related error occurred",	×
112	slog.String("error", err.Error()))	×
113	return errors.New(base.ErrorCode_ERROR_CODE_CANCELLED.String())	×
114	}	×
115
116	// Check if the error is serialization-related
117	if IsSerializationRelatedError(err) {	×
118	slog.DebugContext(ctx, "A serialization-related error occurred",	×
119	slog.String("error", err.Error()))	×
120	return errors.New(base.ErrorCode_ERROR_CODE_SERIALIZATION.String())	×
121	}	×
122
123	// For all other types of errors, log them at the error level and record them in the span
124	slog.ErrorContext(ctx, "An operational error occurred",	×
125	slog.Any("error", err))	×
126	span.RecordError(err)	×
127	span.SetStatus(codes.Error, err.Error())	×
128		×
129	// Return a new error with the standard error code provided	×
130	return errors.New(errorCode.String())	×
131	}
132
133	// IsContextRelatedError checks if the error is due to context cancellation, deadline exceedance, or closed connection
134	func IsContextRelatedError(ctx context.Context, err error) bool {	×
135	if errors.Is(ctx.Err(), context.Canceled) \|\| errors.Is(ctx.Err(), context.DeadlineExceeded) {	×
136	return true	×
137	}	×
138	if errors.Is(err, context.Canceled) \|\|	×
139	errors.Is(err, context.DeadlineExceeded) \|\|	×
140	strings.Contains(err.Error(), "conn closed") {	×
141	return true	×
142	}	×
143	return false	×
144	}
145
146	// IsSerializationRelatedError checks if the error is a serialization failure, typically in database transactions.
147	func IsSerializationRelatedError(err error) bool {	×
148	if strings.Contains(err.Error(), "could not serialize") \|\|	×
149	strings.Contains(err.Error(), "duplicate key value") {	×
150	return true	×
151	}	×
152	return false	×
153	}
154
155	// WaitWithBackoff implements an exponential backoff strategy with jitter for retries.
156	// It waits for a calculated duration or until the context is cancelled, whichever comes first.
157	func WaitWithBackoff(ctx context.Context, tenantID string, retries int) {	×
158	backoff := time.Duration(math.Min(float64(20time.Millisecond)math.Pow(2, float64(retries)), float64(1*time.Second)))	×
159	jitter := time.Duration(rand.Float64() * float64(backoff) * 0.5)	×
160	nextBackoff := backoff + jitter	×
161	slog.WarnContext(ctx, "waiting before retry", slog.String("tenant_id", tenantID), slog.Int64("backoff_duration", nextBackoff.Milliseconds()))	×
162	select {	×
163	case <-time.After(nextBackoff):	×
164	case <-ctx.Done():	×
165	}
166	}

Permify / permify / 10924732005

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous