feat(compass): add max_concurrency config and skip edge-only entity upserts

Add max_concurrency option to limit concurrent requests per batch,
preventing Compass from being overwhelmed during large extractions.

Skip the entity upsert POST for records that carry only edges (no name,
description, or properties), reducing unnecessary API calls for
relationship-only data like collaborator permissions.

Author: ravisuhag

plugins/sinks/compass/README.md

@@ -20,15 +20,16 @@ sinks:
 | :-- | :--- | :------ | :---------- | :- |
 | `host` | `string` | `https://compass.com` | Hostname of the Compass service | *required* |
 | `headers` | `map` | `Compass-User-UUID: meteor@raystack.io` | Additional HTTP headers to send with each request. Multiple values are comma-separated. | *optional* |
+| `max_concurrency` | `int` | `10` | Maximum concurrent requests per batch. `0` means no limit (default). | *optional* |
 
 ## Behavior
 
 For each Record the sink:
 
-1. **Upserts the Entity** via `POST /raystack.compass.v1beta1.CompassService/UpsertEntity` -- sends `urn`, `type`, `name`, `description`, `source`, and `properties`.
+1. **Upserts the Entity** via `POST /raystack.compass.v1beta1.CompassService/UpsertEntity` -- sends `urn`, `type`, `name`, `description`, `source`, and `properties`. Skipped for edge-only records (no name, description, or properties).
 2. **Upserts all Edges** via `POST /raystack.compass.v1beta1.CompassService/UpsertEdge` -- one request per edge (e.g. `owned_by`, `derived_from`, `generates`), sending `source_urn`, `target_urn`, `type`, `source`, and `properties`.
 
-Requests are made concurrently within a batch. Server errors (5xx) are returned as retryable errors.
+Requests are made concurrently within a batch (controlled by `max_concurrency`). Server errors (5xx) are returned as retryable errors.
 
 ## Contributing
 

plugins/sinks/compass/sink.go

@@ -24,8 +24,9 @@ import (
 var summary string
 
 type Config struct {
-	Host    string            `json:"host" yaml:"host" mapstructure:"host" validate:"required"`
-	Headers map[string]string `json:"headers" yaml:"headers" mapstructure:"headers"`
+	Host           string            `json:"host" yaml:"host" mapstructure:"host" validate:"required"`
+	Headers        map[string]string `json:"headers" yaml:"headers" mapstructure:"headers"`
+	MaxConcurrency int               `json:"max_concurrency" yaml:"max_concurrency" mapstructure:"max_concurrency"`
 }
 
 var info = plugins.Info{
@@ -39,6 +40,8 @@ var info = plugins.Info{
 	headers:
 	  Compass-User-UUID: meteor@raystack.io
 	  X-Other-Header: value1, value2
+	# Maximum number of concurrent requests to compass per batch. 0 means no limit.
+	# max_concurrency: 10
 	`),
 }
 
@@ -84,7 +87,11 @@ func (s *Sink) Sink(ctx context.Context, batch []models.Record) error {
 	}
 
 	errGroup := errgroup.Group{}
-	errGroup.SetLimit(len(batch))
+	limit := len(batch)
+	if s.config.MaxConcurrency > 0 && s.config.MaxConcurrency < limit {
+		limit = s.config.MaxConcurrency
+	}
+	errGroup.SetLimit(limit)
 
 	for _, record := range batch {
 		record := record
@@ -109,13 +116,17 @@ func (*Sink) Close() error { return nil }
 func (s *Sink) sinkRecord(ctx context.Context, record models.Record) error {
 	entity := record.Entity()
 
-	// 1. Upsert the entity.
-	entityReq := s.buildEntityRequest(record)
-	if err := s.post(ctx, upsertEntityRoute, entityReq); err != nil {
-		return fmt.Errorf("upsert entity: %w", err)
+	// Skip entity upsert only for bare records that exist solely to carry edges
+	// (no properties, no name, no description — just a URN and type for edge context).
+	edgeOnly := entity.GetProperties() == nil && entity.GetName() == "" && entity.GetDescription() == ""
+	if !edgeOnly {
+		entityReq := s.buildEntityRequest(record)
+		if err := s.post(ctx, upsertEntityRoute, entityReq); err != nil {
+			return fmt.Errorf("upsert entity: %w", err)
+		}
 	}
 
-	// 2. Upsert all edges uniformly via UpsertEdge.
+	// Upsert all edges uniformly via UpsertEdge.
 	for _, edge := range record.Edges() {
 		edgeReq := UpsertEdgeRequest{
 			SourceURN: edge.GetSourceUrn(),

plugins/sinks/compass/sink_test.go

@@ -73,7 +73,7 @@ func TestSink(t *testing.T) {
 		}})
 		require.NoError(t, err)
 
-		entity := &meteorv1beta1.Entity{Type: "table"}
+		entity := &meteorv1beta1.Entity{Type: "table", Name: "my-table"}
 		err = compassSink.Sink(ctx, []models.Record{models.NewRecord(entity)})
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "compass returns 404")
@@ -92,7 +92,7 @@ func TestSink(t *testing.T) {
 				}})
 				require.NoError(t, err)
 
-				entity := &meteorv1beta1.Entity{Type: "table"}
+				entity := &meteorv1beta1.Entity{Type: "table", Name: "my-table"}
 				err = compassSink.Sink(ctx, []models.Record{models.NewRecord(entity)})
 				require.Error(t, err)
 				assert.ErrorAs(t, err, &plugins.RetryError{})
@@ -305,6 +305,67 @@ func TestSink(t *testing.T) {
 		assert.Contains(t, req.Header.Get("Compass-User-UUID"), "meteor@raystack.io")
 	})
 
+	t.Run("should skip entity upsert for edge-only records", func(t *testing.T) {
+		client := &mockHTTPClient{}
+		client.SetupResponse(200, `{}`)
+		ctx := context.TODO()
+
+		compassSink := compass.New(client, testutils.Logger)
+		err := compassSink.Init(ctx, plugins.Config{RawConfig: map[string]any{
+			"host": host,
+		}})
+		require.NoError(t, err)
+
+		// Entity with only URN/type/source — no name, description, or properties.
+		entity := &meteorv1beta1.Entity{
+			Urn:    "my-repo-urn",
+			Source: "github",
+			Type:   "repository",
+		}
+		edges := []*meteorv1beta1.Edge{
+			{SourceUrn: "user-urn-1", TargetUrn: "my-repo-urn", Type: "has_access_to", Source: "github"},
+			{SourceUrn: "user-urn-2", TargetUrn: "my-repo-urn", Type: "has_access_to", Source: "github"},
+		}
+		err = compassSink.Sink(ctx, []models.Record{models.NewRecord(entity, edges...)})
+		assert.NoError(t, err)
+
+		// Should have only 2 edge requests, no entity upsert.
+		require.Len(t, client.requests, 2)
+		assert.Equal(t, upsertEdgeURL, reqURL(client.requests[0]))
+		assert.Equal(t, upsertEdgeURL, reqURL(client.requests[1]))
+	})
+
+	t.Run("should respect max_concurrency config", func(t *testing.T) {
+		client := &mockHTTPClient{}
+		client.SetupResponse(200, `{}`)
+		ctx := context.TODO()
+
+		compassSink := compass.New(client, testutils.Logger)
+		err := compassSink.Init(ctx, plugins.Config{RawConfig: map[string]any{
+			"host":            host,
+			"max_concurrency": 2,
+		}})
+		require.NoError(t, err)
+
+		// Create a batch of 5 records.
+		var batch []models.Record
+		for i := 0; i < 5; i++ {
+			entity := &meteorv1beta1.Entity{
+				Urn:    fmt.Sprintf("urn-%d", i),
+				Name:   fmt.Sprintf("entity-%d", i),
+				Source: "test",
+				Type:   "table",
+			}
+			batch = append(batch, models.NewRecord(entity))
+		}
+
+		err = compassSink.Sink(ctx, batch)
+		assert.NoError(t, err)
+
+		// All 5 entity requests should complete despite concurrency limit.
+		require.Len(t, client.requests, 5)
+	})
+
 	t.Run("should flatten data into properties without @type", func(t *testing.T) {
 		client := &mockHTTPClient{}
 		client.SetupResponse(200, `{}`)