Repository: xataio/pgstream
Pull request #110 (2025-01-17 14:55:13)

Fix validation error parsing on opensearch
When the `SendBulkRequest` opensearch/elasticsearch client returns an error, it's not being parsed like the rest of the client methods, which ends up treating all error status as critical. 
This PR updates the handling, so that validation query errors don't stop the processing of the search batch indexer.


Patch:
diff --git a/internal/searchstore/elasticsearch/elasticsearch_client.go b/internal/searchstore/elasticsearch/elasticsearch_client.go
index db8c697..f870958 100644
--- a/internal/searchstore/elasticsearch/elasticsearch_client.go
+++ b/internal/searchstore/elasticsearch/elasticsearch_client.go
@@ -422,13 +422,15 @@ func (ec *Client) SendBulkRequest(ctx context.Context, items []searchstore.BulkI
 		return nil, fmt.Errorf("perform: %w", err)
 	}
 	defer resp.Body.Close()
+
+	if resp.StatusCode > 299 {
+		return nil, fmt.Errorf("[SendBulkRequest] error response from Elasticsearch: %w", searchstore.ExtractResponseError(resp.Body, resp.StatusCode))
+	}
+
 	bodyBytes, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, fmt.Errorf("read body: %w", err)
 	}
-	if resp.StatusCode > 299 {
-		return nil, fmt.Errorf("error from Elasticsearch: %d: %s", resp.StatusCode, bodyBytes)
-	}
 
 	return searchstore.VerifyResponse(bodyBytes, items)
 }
diff --git a/internal/searchstore/opensearch/opensearch_client.go b/internal/searchstore/opensearch/opensearch_client.go
index ab64c90..c9bc577 100644
--- a/internal/searchstore/opensearch/opensearch_client.go
+++ b/internal/searchstore/opensearch/opensearch_client.go
@@ -422,13 +422,15 @@ func (c *Client) SendBulkRequest(ctx context.Context, items []searchstore.BulkIt
 		return nil, fmt.Errorf("perform: %w", err)
 	}
 	defer resp.Body.Close()
+
+	if resp.StatusCode > 299 {
+		return nil, fmt.Errorf("[SendBulkRequest] error response from OpenSearch: %w", searchstore.ExtractResponseError(resp.Body, resp.StatusCode))
+	}
+
 	bodyBytes, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, fmt.Errorf("read body: %w", err)
 	}
-	if resp.StatusCode > 299 {
-		return nil, fmt.Errorf("error from OpenSearch: %d: %s", resp.StatusCode, bodyBytes)
-	}
 
 	return searchstore.VerifyResponse(bodyBytes, items)
 }
diff --git a/internal/searchstore/search_errors.go b/internal/searchstore/search_errors.go
index ecbbe8e..5385818 100644
--- a/internal/searchstore/search_errors.go
+++ b/internal/searchstore/search_errors.go
@@ -96,17 +96,17 @@ type apiResponse interface {
 func IsErrResponse(res apiResponse) error {
 	if res.IsError() {
 		if res.GetStatusCode() == http.StatusNotFound {
-			return fmt.Errorf("%w: %w", ErrResourceNotFound, extractResponseError(res))
+			return fmt.Errorf("%w: %w", ErrResourceNotFound, ExtractResponseError(res.GetBody(), res.GetStatusCode()))
 		}
-		return extractResponseError(res)
+		return ExtractResponseError(res.GetBody(), res.GetStatusCode())
 	}
 
 	return nil
 }
 
-func extractResponseError(res apiResponse) error {
+func ExtractResponseError(body io.ReadCloser, statusCode int) error {
 	var e map[string]any
-	if err := json.NewDecoder(res.GetBody()).Decode(&e); err != nil {
+	if err := json.NewDecoder(body).Decode(&e); err != nil {
 		return fmt.Errorf("decoding error response: %w", err)
 	}
 
@@ -150,7 +150,6 @@ func extractResponseError(res apiResponse) error {
 		}
 	}
 
-	statusCode := res.GetStatusCode()
 	if err, ok := getRetryableError(statusCode); ok {
 		return RetryableError{Cause: err}
 	}
diff --git a/pkg/wal/processor/search/errors.go b/pkg/wal/processor/search/errors.go
index 4daccb6..541b758 100644
--- a/pkg/wal/processor/search/errors.go
+++ b/pkg/wal/processor/search/errors.go
@@ -47,7 +47,8 @@ func (e ErrSchemaUpdateOutOfOrder) Error() string {
 }
 
 var (
-	ErrRetriable = errors.New("retriable error")
+	ErrRetriable    = errors.New("retriable error")
+	ErrInvalidQuery = errors.New("invalid query")
 
 	errNilIDValue      = errors.New("id has nil value")
 	errNilVersionValue = errors.New("version has nil value")
diff --git a/pkg/wal/processor/search/search_batch_indexer.go b/pkg/wal/processor/search/search_batch_indexer.go
index 3105ed9..cc6b9f1 100644
--- a/pkg/wal/processor/search/search_batch_indexer.go
+++ b/pkg/wal/processor/search/search_batch_indexer.go
@@ -256,7 +256,12 @@ func (i *BatchIndexer) sendBatch(ctx context.Context, batch *msgBatch) error {
 		if len(writes) > 0 {
 			failed, err := i.store.SendDocuments(ctx, writes)
 			if err != nil {
-				return err
+				i.logger.Error(err, "search store error when sending documents", loglib.Fields{
+					"documents": writes,
+				})
+				if !errors.Is(err, ErrInvalidQuery) {
+					return err
+				}
 			}
 			if len(failed) > 0 {
 				i.logger.Error(nil, "failed to send documents", loglib.Fields{
diff --git a/pkg/wal/processor/search/store/search_store.go b/pkg/wal/processor/search/store/search_store.go
index a283da3..8021771 100644
--- a/pkg/wal/processor/search/store/search_store.go
+++ b/pkg/wal/processor/search/store/search_store.go
@@ -469,7 +469,10 @@ func (s *Store) ensureSchemaMapping(ctx context.Context, schemaName string, meta
 
 func mapError(err error) error {
 	if errors.As(err, &searchstore.RetryableError{}) {
-		return fmt.Errorf("%w: %v", search.ErrRetriable, err.Error())
+		return fmt.Errorf("%w: %w", search.ErrRetriable, err)
+	}
+	if errors.As(err, &searchstore.ErrQueryInvalid{}) {
+		return fmt.Errorf("%w: %w", search.ErrInvalidQuery, err)
 	}
 	return err
 }


Test patch:
diff --git a/pkg/wal/processor/search/search_batch_indexer_test.go b/pkg/wal/processor/search/search_batch_indexer_test.go
index c940ccf..b5840b1 100644
--- a/pkg/wal/processor/search/search_batch_indexer_test.go
+++ b/pkg/wal/processor/search/search_batch_indexer_test.go
@@ -635,6 +635,23 @@ func TestBatchIndexer_sendBatch(t *testing.T) {
 
 			wantErr: errTest,
 		},
+		{
+			name: "error - sending documents with validation failure",
+			batch: &msgBatch{
+				msgs: []*msg{
+					{write: testDocument1},
+					{write: testDocument2},
+				},
+				positions: []wal.CommitPosition{testCommitPos},
+			},
+			store: &mockStore{
+				sendDocumentsFn: func(ctx context.Context, _ uint, docs []Document) ([]DocumentError, error) {
+					return nil, ErrInvalidQuery
+				},
+			},
+
+			wantErr: nil,
+		},
 		{
 			name: "error - checkpointing",
 			batch: &msgBatch{