这是indexloc提供的服务，不要输入任何密码
Skip to content

feat: bulk checker goroutine cancellation #1694

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Oct 16, 2024
Merged

feat: bulk checker goroutine cancellation #1694

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0ddc62d
feat(bulk-checker): add goroutine cancellation on success limit and i…
tolgaozen Oct 16, 2024
6f57b22
feat: default page size for data read apis
tolgaozen Oct 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .goreleaser.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ nfpms:
maintainer: "permify <help@permify.co>"
vendor: "permify inc."
snapshot:
name_template: "{{ incpatch .Version }}-next"
version_template: "{{ incpatch .Version }}-next"
Copy link

@coderabbitai coderabbitai bot Oct 16, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Codebase verification

Issues with the change from name_template to version_template

The version_template key was not found in GoReleaser's documentation for snapshot configuration. Additionally, name_template is still present in .goreleaser.yml. Please verify and correct the configuration to ensure it aligns with GoReleaser's requirements.

🔗 Analysis chain

Confirm the change from name_template to version_template

The change from name_template to version_template in the snapshot section appears to be intentional. This modification alters how versioning for snapshots is defined.

To ensure this change aligns with GoReleaser's current best practices, please run the following verification script:

This script will help verify if the change is consistent with GoReleaser's current documentation and if all relevant occurrences have been updated.

🏁 Scripts executed

The following scripts were executed for the analysis:

Script:

#!/bin/bash
# Description: Verify if 'version_template' is the correct key for snapshot configuration in GoReleaser

# Test: Check GoReleaser's documentation for the correct key in snapshot configuration
curl -s https://goreleaser.com/customization/snapshot/ | grep -q 'version_template:'
if [ $? -eq 0 ]; then
    echo "The change to 'version_template' is correct according to GoReleaser's documentation."
else
    echo "Warning: 'version_template' was not found in GoReleaser's documentation for snapshot configuration. Please double-check this change."
fi

# Test: Check for any remaining occurrences of 'name_template' in the snapshot section
if grep -q 'name_template' .goreleaser.yml; then
    echo "Warning: 'name_template' is still present in .goreleaser.yml. Ensure all occurrences have been updated if intended."
else
    echo "All occurrences of 'name_template' have been updated in .goreleaser.yml."
fi

Length of output: 673

changelog:
sort: asc
filters:
Expand Down
2 changes: 1 addition & 1 deletion docs/api-reference/apidocs.swagger.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
"info": {
"title": "Permify API",
"description": "Permify is an open source authorization service for creating fine-grained and scalable authorization systems.",
"version": "v1.1.3",
"version": "v1.1.4",
"contact": {
"name": "API Support",
"url": "https://github.com/Permify/permify/issues",
Expand Down
2 changes: 1 addition & 1 deletion docs/api-reference/openapiv2/apidocs.swagger.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
"info": {
"title": "Permify API",
"description": "Permify is an open source authorization service for creating fine-grained and scalable authorization systems.",
"version": "v1.1.3",
"version": "v1.1.4",
"contact": {
"name": "API Support",
"url": "https://github.com/Permify/permify/issues",
Expand Down
60 changes: 50 additions & 10 deletions internal/engines/bulk.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ import (
"sync"
"sync/atomic"

"github.com/pkg/errors"

Comment on lines +9 to +10
Copy link

@coderabbitai coderabbitai bot Oct 16, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Consider using the standard errors package

Since Go 1.13, the standard library's errors package provides errors.Is and errors.As functions. You may not need to import "github.com/pkg/errors".

Apply this diff to switch to the standard library:

-	"github.com/pkg/errors"
+	"errors"
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
"github.com/pkg/errors"
"errors"

"golang.org/x/sync/errgroup"
"golang.org/x/sync/semaphore"

Expand Down Expand Up @@ -123,6 +125,15 @@ func (bc *BulkChecker) CollectAndSortRequests() {
}
}

// prepareRequestList sorts and prepares a copy of the request list.
func (bc *BulkChecker) prepareRequestList() []BulkCheckerRequest {
bc.mu.Lock()
defer bc.mu.Unlock()

bc.sortRequests() // Sort requests based on ID
return append([]BulkCheckerRequest{}, bc.list...) // Return a copy of the list to avoid modifying the original
}

// StopCollectingRequests Signal to stop collecting requests and close the channel
func (bc *BulkChecker) StopCollectingRequests() {
bc.mu.Lock()
Expand Down Expand Up @@ -152,17 +163,18 @@ func (bc *BulkChecker) ExecuteRequests(size uint32) error {
// Wait for request collection to complete before proceeding
bc.wg.Wait()

// Create a context with cancellation that will be used to stop further processing
ctx, cancel := context.WithCancel(bc.ctx)
defer cancel() // Ensure the context is canceled when done

// Track the number of successful permission checks
successCount := int64(0)
// Semaphore to control the maximum number of concurrent permission checks
sem := semaphore.NewWeighted(int64(bc.concurrencyLimit))
var mu sync.Mutex

// Lock the mutex to prevent race conditions while sorting and copying the list of requests
bc.mu.Lock()
bc.sortRequests() // Sort requests based on id
listCopy := append([]BulkCheckerRequest{}, bc.list...) // Create a copy of the list to avoid modifying the original during processing
bc.mu.Unlock() // Unlock the mutex after sorting and copying
// Sort and copy the list of requests
listCopy := bc.prepareRequestList()

// Pre-allocate a slice to store the results of the permission checks
results := make([]base.CheckResult, len(listCopy))
Expand All @@ -173,6 +185,7 @@ func (bc *BulkChecker) ExecuteRequests(size uint32) error {
for i, currentRequest := range listCopy {
// If we've reached the success limit, stop processing further requests
if atomic.LoadInt64(&successCount) >= int64(size) {
cancel() // Cancel the context to stop further goroutines
break
}

Expand All @@ -181,18 +194,31 @@ func (bc *BulkChecker) ExecuteRequests(size uint32) error {

// Use errgroup to manage the goroutines, which allows for error handling and synchronization
bc.g.Go(func() error {
// Check if the context has been canceled, and return without error if so
if err := ctx.Err(); err == context.Canceled {
return nil // Gracefully exit the goroutine if context is canceled
}

Comment on lines +198 to +201
Copy link

@coderabbitai coderabbitai bot Oct 16, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Handle all context-related errors when checking context

Currently, only context.Canceled is checked. Consider handling context.DeadlineExceeded as well to cover all context cancellation scenarios.

Apply this diff:

-	if err := ctx.Err(); err == context.Canceled {
+	if err := ctx.Err(); err != nil {
		return nil // Gracefully exit the goroutine if context is canceled or deadline exceeded
	}

Alternatively, use the IsContextRelatedError function:

-	if err := ctx.Err(); err == context.Canceled {
+	if IsContextRelatedError(ctx, ctx.Err()) {
		return nil // Gracefully exit the goroutine if context is canceled or deadline exceeded
	}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
if err := ctx.Err(); err == context.Canceled {
return nil // Gracefully exit the goroutine if context is canceled
}
if err := ctx.Err(); err != nil {
return nil // Gracefully exit the goroutine if context is canceled or deadline exceeded
}

// Acquire a slot in the semaphore to control concurrency
if err := sem.Acquire(bc.ctx, 1); err != nil {
return err // Return an error if semaphore acquisition fails
if err := sem.Acquire(ctx, 1); err != nil {
// Return nil instead of error if the context was canceled during semaphore acquisition
if IsContextRelatedError(ctx, err) {
return nil
}
return err // Return an error if semaphore acquisition fails for other reasons
}
defer sem.Release(1) // Ensure the semaphore slot is released after processing

var result base.CheckResult
if req.Result == base.CheckResult_CHECK_RESULT_UNSPECIFIED {
// Perform the permission check if the result is not already specified
cr, err := bc.checker.Check(bc.ctx, req.Request)
cr, err := bc.checker.Check(ctx, req.Request)
if err != nil {
return err // Return an error if the check fails
// Handle context cancellation error here
if IsContextRelatedError(ctx, err) {
return nil // Ignore the cancellation error
}
return err // Return the actual error if it's not due to cancellation
}
result = cr.GetCan() // Get the result from the check
} else {
Expand Down Expand Up @@ -230,13 +256,22 @@ func (bc *BulkChecker) ExecuteRequests(size uint32) error {
}
mu.Unlock() // Unlock the mutex after updating the results and processed index

// If the success limit has been reached, cancel the context to stop further processing
if atomic.LoadInt64(&successCount) >= int64(size) {
cancel()
}

return nil // Return nil to indicate successful processing
})
}

// Wait for all goroutines to complete and check for any errors
if err := bc.g.Wait(); err != nil {
return err // Return the error if any goroutine returned an error
// Ignore context cancellation as an error
if IsContextRelatedError(ctx, err) {
return nil
}
return err // Return other errors if any goroutine returned an error
}

return nil // Return nil if all processing completed successfully
Expand Down Expand Up @@ -307,3 +342,8 @@ func (s *BulkSubjectPublisher) Publish(subject *base.Subject, metadata *base.Per
Result: result,
}
}

// IsContextRelatedError checks if the error is due to context cancellation, deadline exceedance, or closed connection
func IsContextRelatedError(ctx context.Context, err error) bool {
return errors.Is(ctx.Err(), context.Canceled) || errors.Is(ctx.Err(), context.DeadlineExceeded)
}
Comment on lines +347 to +349
Copy link

@coderabbitai coderabbitai bot Oct 16, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Fix IsContextRelatedError to check the passed error

The function currently checks ctx.Err() instead of the provided err. It should check err to accurately determine if the error is context-related.

Apply this diff:

func IsContextRelatedError(ctx context.Context, err error) bool {
-	return errors.Is(ctx.Err(), context.Canceled) || errors.Is(ctx.Err(), context.DeadlineExceeded)
+	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
func IsContextRelatedError(ctx context.Context, err error) bool {
return errors.Is(ctx.Err(), context.Canceled) || errors.Is(ctx.Err(), context.DeadlineExceeded)
}
func IsContextRelatedError(ctx context.Context, err error) bool {
return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
}

53 changes: 23 additions & 30 deletions internal/engines/entityFilter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,9 @@ func (engine *EntityFilter) attributeEntrance(
g *errgroup.Group, // An errgroup used for executing goroutines.
publisher *BulkEntityPublisher, // A custom publisher that publishes results in bulk.
) error { // Returns an error if one occurs during execution.
if !visits.AddEA(entrance.TargetEntrance.GetType(), entrance.TargetEntrance.GetValue()) { // If the entity and relation has already been visited.
return nil
}

// Retrieve the scope associated with the target entrance type.
// Check if it exists to avoid accessing a nil map entry.
Expand Down Expand Up @@ -180,21 +183,31 @@ func (engine *EntityFilter) attributeEntrance(
// NewUniqueTupleIterator() ensures that the iterator only returns unique tuples.
it := database.NewUniqueAttributeIterator(rit, cti)

for it.HasNext() { // Loop over each relationship.
// Get the next tuple's subject.
// Iterate over the relationships.
for it.HasNext() {
// Get the next attribute's entity.
current, ok := it.GetNext()

if !ok {
break
}

g.Go(func() error {
return engine.la(ctx, request,
&base.Entity{
Type: entrance.TargetEntrance.GetType(),
Id: current.GetEntity().GetId(),
}, visits, g, publisher)
})
// Extract the entity details.
entity := &base.Entity{
Type: entrance.TargetEntrance.GetType(), // Example: using the type from a previous variable 'entrance'
Id: current.GetEntity().GetId(),
}

// Check if the entity has already been visited to prevent processing it again.
if !visits.AddPublished(entity) {
continue // Skip this entity if it has already been visited.
}

// Publish the entity with its metadata.
publisher.Publish(entity, &base.PermissionCheckRequestMetadata{
SnapToken: request.GetMetadata().GetSnapToken(),
SchemaVersion: request.GetMetadata().GetSchemaVersion(),
Depth: request.GetMetadata().GetDepth(),
}, request.GetContext(), base.CheckResult_CHECK_RESULT_UNSPECIFIED)
}

return nil
Expand Down Expand Up @@ -443,23 +456,3 @@ func (engine *EntityFilter) lt(
})
return nil
}

func (engine *EntityFilter) la(
ctx context.Context, // A context used for tracing and cancellation.
request *base.PermissionEntityFilterRequest, // A permission request for linked entities.
found *base.Entity, // An entity and relation that was previously found.
visits *VisitsMap, // A map that keeps track of visited entities to avoid infinite loops.
g *errgroup.Group, // An errgroup used for executing goroutines.
publisher *BulkEntityPublisher, // A custom publisher that publishes results in bulk.
) error { // Returns an error if one occurs during execution.
if !visits.AddPublished(found) { // If the entity and relation has already been visited.
return nil
}

publisher.Publish(found, &base.PermissionCheckRequestMetadata{ // Publish the found entity with the permission check metadata.
SnapToken: request.GetMetadata().GetSnapToken(),
SchemaVersion: request.GetMetadata().GetSchemaVersion(),
Depth: request.GetMetadata().GetDepth(),
}, request.GetContext(), base.CheckResult_CHECK_RESULT_UNSPECIFIED)
return nil
}
7 changes: 7 additions & 0 deletions internal/engines/utils.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ type CheckResponse struct {
// VisitsMap - a thread-safe map of ENR records.
type VisitsMap struct {
er sync.Map
ea sync.Map
published sync.Map
}

Expand All @@ -91,6 +92,12 @@ func (s *VisitsMap) AddER(entity *base.Entity, relation string) bool {
return !existed
}

func (s *VisitsMap) AddEA(entityType, attribute string) bool {
key := fmt.Sprintf("%s$%s", entityType, attribute)
_, existed := s.er.LoadOrStore(key, struct{}{})
return !existed
}
Comment on lines +95 to +99
Copy link

@coderabbitai coderabbitai bot Oct 16, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Fix the map used in AddEA method.

The AddEA method is using the er map instead of the newly added ea map. This appears to be an error and should be corrected to ensure proper functionality.

Please apply the following fix:

 func (s *VisitsMap) AddEA(entityType, attribute string) bool {
 	key := fmt.Sprintf("%s$%s", entityType, attribute)
-	_, existed := s.er.LoadOrStore(key, struct{}{})
+	_, existed := s.ea.LoadOrStore(key, struct{}{})
 	return !existed
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
func (s *VisitsMap) AddEA(entityType, attribute string) bool {
key := fmt.Sprintf("%s$%s", entityType, attribute)
_, existed := s.er.LoadOrStore(key, struct{}{})
return !existed
}
func (s *VisitsMap) AddEA(entityType, attribute string) bool {
key := fmt.Sprintf("%s$%s", entityType, attribute)
_, existed := s.ea.LoadOrStore(key, struct{}{})
return !existed
}


func (s *VisitsMap) AddPublished(entity *base.Entity) bool {
key := tuple.EntityToString(entity)
_, existed := s.published.LoadOrStore(key, struct{}{})
Expand Down
2 changes: 1 addition & 1 deletion internal/info.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ var Identifier = ""
*/
const (
// Version is the last release of the Permify (e.g. v0.1.0)
Version = "v1.1.3"
Version = "v1.1.4"
)

// Function to create a single line of the ASCII art with centered content and color
Expand Down
14 changes: 12 additions & 2 deletions internal/servers/dataServer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,11 @@ func (r *DataServer) ReadRelationships(ctx context.Context, request *v1.Relation
defer span.End()
start := time.Now()

size := request.GetPageSize()
if size == 0 {
size = 50
}

v := request.Validate()
if v != nil {
return nil, status.Error(GetStatus(v), v.Error())
Expand All @@ -83,7 +88,7 @@ func (r *DataServer) ReadRelationships(ctx context.Context, request *v1.Relation
request.GetFilter(),
snap,
database.NewPagination(
database.Size(request.GetPageSize()),
database.Size(size),
database.Token(request.GetContinuousToken()),
),
)
Expand All @@ -109,6 +114,11 @@ func (r *DataServer) ReadAttributes(ctx context.Context, request *v1.AttributeRe
defer span.End()
start := time.Now()

size := request.GetPageSize()
if size == 0 {
size = 50
}

v := request.Validate()
if v != nil {
return nil, status.Error(GetStatus(v), v.Error())
Expand All @@ -129,7 +139,7 @@ func (r *DataServer) ReadAttributes(ctx context.Context, request *v1.AttributeRe
request.GetFilter(),
snap,
database.NewPagination(
database.Size(request.GetPageSize()),
database.Size(size),
database.Token(request.GetContinuousToken()),
),
)
Expand Down
2 changes: 1 addition & 1 deletion pkg/pb/base/v1/openapi.pb.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion proto/base/v1/openapi.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_swagger) = {
info: {
title: "Permify API";
description: "Permify is an open source authorization service for creating fine-grained and scalable authorization systems.";
version: "v1.1.3";
version: "v1.1.4";
contact: {
name: "API Support";
url: "https://github.com/Permify/permify/issues";
Expand Down