Delete opensearch schema synchronously

README.md

@@ -176,23 +176,18 @@ One of exponential/constant backoff policies can be provided for the Kafka commi
 <details>
   <summary>Search Batch Indexer</summary>
 
-| Environment Variable                                         | Default | Required | Description                                                                                                    |
-| ------------------------------------------------------------ | ------- | -------- | -------------------------------------------------------------------------------------------------------------- |
-| PGSTREAM_OPENSEARCH_STORE_URL                                | N/A     | Yes      | URL for the opensearch store to connect to (at least one of the URLs must be provided).                        |
-| PGSTREAM_ELASTICSEARCH_STORE_URL                             | N/A     | Yes      | URL for the elasticsearch store to connect to (at least one of the URLs must be provided).                     |
-| PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT                        | 1s      | No       | Max time interval at which the batch sending to the search store is triggered.                                 |
-| PGSTREAM_SEARCH_INDEXER_BATCH_SIZE                           | 100     | No       | Max number of messages to be sent per batch. When this size is reached, the batch is sent to the search store. |
-| PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES                      | 100MiB  | No       | Max memory used by the search batch indexer for inflight batches.                                              |
-| PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_INITIAL_INTERVAL | 0       | No       | Initial interval for the exponential backoff policy to be applied to the search indexer cleanup retries.       |
-| PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_MAX_INTERVAL     | 0       | No       | Max interval for the exponential backoff policy to be applied to the search indexer cleanup retries.           |
-| PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_MAX_RETRIES      | 0       | No       | Max retries for the exponential backoff policy to be applied to the search indexer cleanup retries.            |
-| PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_INTERVAL             | 0       | No       | Constant interval for the backoff policy to be applied to the search indexer cleanup retries.                  |
-| PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_MAX_RETRIES          | 0       | No       | Max retries for the backoff policy to be applied to the search indexer cleanup retries.                        |
-| PGSTREAM_SEARCH_STORE_EXP_BACKOFF_INITIAL_INTERVAL           | 1s      | No       | Initial interval for the exponential backoff policy to be applied to the search store operation retries.       |
-| PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_INTERVAL               | 1min    | No       | Max interval for the exponential backoff policy to be applied to the search store operation retries.           |
-| PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_RETRIES                | 0       | No       | Max retries for the exponential backoff policy to be applied to the search store operation retries.            |
-| PGSTREAM_SEARCH_STORE_BACKOFF_INTERVAL                       | 0       | No       | Constant interval for the backoff policy to be applied to the search store operation retries.                  |
-| PGSTREAM_SEARCH_STORE_BACKOFF_MAX_RETRIES                    | 0       | No       | Max retries for the backoff policy to be applied to the search store operation retries.                        |
+| Environment Variable                               | Default | Required | Description                                                                                                    |
+| -------------------------------------------------- | ------- | -------- | -------------------------------------------------------------------------------------------------------------- | --- |
+| PGSTREAM_OPENSEARCH_STORE_URL                      | N/A     | Yes      | URL for the opensearch store to connect to (at least one of the URLs must be provided).                        |
+| PGSTREAM_ELASTICSEARCH_STORE_URL                   | N/A     | Yes      | URL for the elasticsearch store to connect to (at least one of the URLs must be provided).                     |
+| PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT              | 1s      | No       | Max time interval at which the batch sending to the search store is triggered.                                 |
+| PGSTREAM_SEARCH_INDEXER_BATCH_SIZE                 | 100     | No       | Max number of messages to be sent per batch. When this size is reached, the batch is sent to the search store. |
+| PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES            | 100MiB  | No       | Max memory used by the search batch indexer for inflight batches.                                              |     |
+| PGSTREAM_SEARCH_STORE_EXP_BACKOFF_INITIAL_INTERVAL | 1s      | No       | Initial interval for the exponential backoff policy to be applied to the search store operation retries.       |
+| PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_INTERVAL     | 1min    | No       | Max interval for the exponential backoff policy to be applied to the search store operation retries.           |
+| PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_RETRIES      | 0       | No       | Max retries for the exponential backoff policy to be applied to the search store operation retries.            |
+| PGSTREAM_SEARCH_STORE_BACKOFF_INTERVAL             | 0       | No       | Constant interval for the backoff policy to be applied to the search store operation retries.                  |
+| PGSTREAM_SEARCH_STORE_BACKOFF_MAX_RETRIES          | 0       | No       | Max retries for the backoff policy to be applied to the search store operation retries.                        |
 
 One of exponential/constant backoff policies can be provided for the search indexer cleanup retry strategy. If none is provided, no retries apply.
 

cmd/config.go

@@ -156,10 +156,9 @@ func parseSearchProcessorConfig() *stream.SearchProcessorConfig {
 
 	return &stream.SearchProcessorConfig{
 		Indexer: search.IndexerConfig{
-			BatchSize:      viper.GetInt("PGSTREAM_SEARCH_INDEXER_BATCH_SIZE"),
-			BatchTime:      viper.GetDuration("PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT"),
-			MaxQueueBytes:  viper.GetInt64("PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES"),
-			CleanupBackoff: parseBackoffConfig("PGSTREAM_SEARCH_INDEXER_CLEANUP"),
+			BatchSize:     viper.GetInt("PGSTREAM_SEARCH_INDEXER_BATCH_SIZE"),
+			BatchTime:     viper.GetDuration("PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT"),
+			MaxQueueBytes: viper.GetInt64("PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES"),
 		},
 		Store: store.Config{
 			OpenSearchURL:    opensearchStore,

kafka2os.env

@@ -10,9 +10,6 @@ PGSTREAM_KAFKA_COMMIT_BACKOFF_MAX_RETRIES=60
 # Processor config
 PGSTREAM_SEARCH_INDEXER_BATCH_SIZE=100
 PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT=5s
-PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_INITIAL_INTERVAL=1s
-PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_MAX_INTERVAL=1m
-PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_MAX_RETRIES=60
 PGSTREAM_OPENSEARCH_STORE_URL="http://admin:admin@localhost:9200"
 PGSTREAM_SEARCH_STORE_BACKOFF_INITIAL_INTERVAL=1s
 PGSTREAM_SEARCH_STORE_BACKOFF_MAX_INTERVAL=1m

pg2os.env

@@ -5,9 +5,6 @@ PGSTREAM_POSTGRES_LISTENER_URL="postgres://postgres:postgres@localhost?sslmode=d
 PGSTREAM_TRANSLATOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost?sslmode=disable"
 PGSTREAM_SEARCH_INDEXER_BATCH_SIZE=100
 PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT=5s
-PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_INITIAL_INTERVAL=1s
-PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_MAX_INTERVAL=1m
-PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_MAX_RETRIES=60
 PGSTREAM_OPENSEARCH_STORE_URL="http://admin:admin@localhost:9200"
 PGSTREAM_SEARCH_STORE_EXP_BACKOFF_INITIAL_INTERVAL=1s
 PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_INTERVAL=1m

pkg/wal/processor/search/config.go

@@ -4,8 +4,6 @@ package search
 
 import (
 	"time"
-
-	"github.com/xataio/pgstream/pkg/backoff"
 )
 
 type IndexerConfig struct {
@@ -18,9 +16,6 @@ type IndexerConfig struct {
 	// MaxQueueBytes is the max memory used by the batch indexer for inflight
 	// batches. Defaults to 100MiB
 	MaxQueueBytes int64
-	// CleanupBackoff is the retry policy to follow for the async index
-	// deletion. If no config is provided, no retry policy is applied.
-	CleanupBackoff backoff.Config
 }
 
 const (

pkg/wal/processor/search/helper_test.go

@@ -24,10 +24,11 @@ func (m *mockAdapter) walEventToMsg(e *wal.Event) (*msg, error) {
 type mockStore struct {
 	getMapperFn            func() Mapper
 	applySchemaChangeFn    func(ctx context.Context, le *schemalog.LogEntry) error
-	deleteSchemaFn         func(ctx context.Context, schemaName string) error
+	deleteSchemaFn         func(ctx context.Context, i uint, schemaName string) error
 	deleteTableDocumentsFn func(ctx context.Context, schemaName string, tableIDs []string) error
 	sendDocumentsFn        func(ctx context.Context, i uint, docs []Document) ([]DocumentError, error)
 	sendDocumentsCalls     uint
+	deleteSchemaCalls      uint
 }
 
 func (m *mockStore) GetMapper() Mapper {
@@ -39,7 +40,8 @@ func (m *mockStore) ApplySchemaChange(ctx context.Context, le *schemalog.LogEntr
 }
 
 func (m *mockStore) DeleteSchema(ctx context.Context, schemaName string) error {
-	return m.deleteSchemaFn(ctx, schemaName)
+	m.deleteSchemaCalls++
+	return m.deleteSchemaFn(ctx, m.deleteSchemaCalls, schemaName)
 }
 
 func (m *mockStore) DeleteTableDocuments(ctx context.Context, schemaName string, tableIDs []string) error {
@@ -51,24 +53,6 @@ func (m *mockStore) SendDocuments(ctx context.Context, docs []Document) ([]Docum
 	return m.sendDocumentsFn(ctx, m.sendDocumentsCalls, docs)
 }
 
-type mockCleaner struct {
-	deleteSchemaFn func(context.Context, string) error
-	startFn        func(context.Context)
-	stopFn         func()
-}
-
-func (m *mockCleaner) deleteSchema(ctx context.Context, schema string) error {
-	return m.deleteSchemaFn(ctx, schema)
-}
-
-func (m *mockCleaner) start(ctx context.Context) {
-	m.startFn(ctx)
-}
-
-func (m *mockCleaner) stop() {
-	m.stopFn()
-}
-
 const (
 	testSchemaName = "test_schema"
 	testTableName  = "test_table"

pkg/wal/processor/search/search_batch_indexer.go

@@ -38,8 +38,6 @@ type BatchIndexer struct {
 
 	// checkpoint callback to mark what was safely stored
 	checkpoint checkpointer.Checkpoint
-
-	cleaner cleaner
 }
 
 type Option func(*BatchIndexer)
@@ -66,10 +64,6 @@ func NewBatchIndexer(ctx context.Context, config IndexerConfig, store Store, lsn
 		opt(indexer)
 	}
 
-	indexer.cleaner = newSchemaCleaner(&config.CleanupBackoff, store, indexer.logger)
-	// start a goroutine for processing schema deletes asynchronously.
-	// routine ends when the internal channel is closed.
-	go indexer.cleaner.start(ctx)
 	return indexer
 }
 
@@ -191,7 +185,6 @@ func (i *BatchIndexer) Name() string {
 
 func (i *BatchIndexer) Close() error {
 	close(i.msgChan)
-	i.cleaner.stop()
 	return nil
 }
 
@@ -270,8 +263,8 @@ func (i *BatchIndexer) applySchemaChange(ctx context.Context, new *schemalog.Log
 	}
 
 	if new.Schema.Dropped {
-		if err := i.cleaner.deleteSchema(ctx, new.SchemaName); err != nil {
-			return fmt.Errorf("register schema for delete: %w", err)
+		if err := i.store.DeleteSchema(ctx, new.SchemaName); err != nil {
+			return fmt.Errorf("deleting schema: %w", err)
 		}
 		return nil
 	}

pkg/wal/processor/search/search_batch_indexer_test.go

@@ -401,7 +401,6 @@ func TestBatchIndexer_sendBatch(t *testing.T) {
 		checkpoint checkpointer.Checkpoint
 		batch      *msgBatch
 		skipSchema func(string) bool
-		cleaner    cleaner
 
 		wantErr error
 	}{
@@ -517,9 +516,8 @@ func TestBatchIndexer_sendBatch(t *testing.T) {
 				},
 				positions: []wal.CommitPosition{testCommitPos},
 			},
-			store: &mockStore{},
-			cleaner: &mockCleaner{
-				deleteSchemaFn: func(ctx context.Context, s string) error {
+			store: &mockStore{
+				deleteSchemaFn: func(ctx context.Context, _ uint, s string) error {
 					require.Equal(t, testSchemaName, s)
 					return nil
 				},
@@ -557,9 +555,8 @@ func TestBatchIndexer_sendBatch(t *testing.T) {
 				},
 				positions: []wal.CommitPosition{testCommitPos},
 			},
-			store: &mockStore{},
-			cleaner: &mockCleaner{
-				deleteSchemaFn: func(ctx context.Context, s string) error {
+			store: &mockStore{
+				deleteSchemaFn: func(ctx context.Context, _ uint, s string) error {
 					return errTest
 				},
 			},
@@ -680,10 +677,6 @@ func TestBatchIndexer_sendBatch(t *testing.T) {
 				indexer.skipSchema = tc.skipSchema
 			}
 
-			if tc.cleaner != nil {
-				indexer.cleaner = tc.cleaner
-			}
-
 			err := indexer.sendBatch(context.Background(), tc.batch)
 			require.ErrorIs(t, err, tc.wantErr)
 		})