fix(search): engine wrapper which handles panics

services/search/pkg/command/server.go

@@ -100,6 +100,9 @@ func Server(cfg *config.Config) *cobra.Command {
 				return fmt.Errorf("unknown search engine: %s", cfg.Engine.Type)
 			}
 
+			// a wrapper for the engine that recovers from panics
+			eng = search.NewRecoveringEngine(eng, logger)
+
 			// initialize gateway selector
 			selector, err := pool.GatewaySelector(cfg.Reva.Address, pool.WithRegistry(registry.GetRegistry()), pool.WithTracerProvider(traceProvider))
 			if err != nil {

services/search/pkg/search/recovering.go

@@ -0,0 +1,131 @@
+package search
+
+import (
+	"context"
+	"fmt"
+	"runtime/debug"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+)
+
+// recoveringEngine guards the engine calls that read from or write to the index.
+// Calls that cannot reach the index are promoted from the wrapped Engine.
+type recoveringEngine struct {
+	Engine
+
+	log log.Logger
+}
+
+// NewRecoveringEngine wraps engine so that panics surface as errors instead of
+// terminating the process
+func NewRecoveringEngine(engine Engine, logger log.Logger) Engine {
+	return &recoveringEngine{Engine: engine, log: logger}
+}
+
+// recoverPanic converts a panic into errp and logs it with the stack,
+// this function should be used always with defer
+func recoverPanic(logger log.Logger, op string, errp *error) {
+	if r := recover(); r != nil {
+		logger.Error().
+			Str("op", op).
+			Interface("panic", r).
+			Str("stack", string(debug.Stack())).
+			Msg("recovered panic from the search engine")
+
+		*errp = fmt.Errorf("%s: search engine panic: %v", op, r)
+	}
+}
+
+func (e *recoveringEngine) Search(ctx context.Context, req *searchService.SearchIndexRequest) (res *searchService.SearchIndexResponse, err error) {
+	defer recoverPanic(e.log, "Search", &err)
+
+	return e.Engine.Search(ctx, req)
+}
+
+// the engine level mutations are guarded as well: the backend builds its own
+// batch internally, which is not wrapped by recoveringBatch
+func (e *recoveringEngine) Upsert(id string, r Resource) (err error) {
+	defer recoverPanic(e.log, "Upsert", &err)
+
+	return e.Engine.Upsert(id, r)
+}
+
+func (e *recoveringEngine) Move(id string, parentID string, targetPath string) (err error) {
+	defer recoverPanic(e.log, "Move", &err)
+
+	return e.Engine.Move(id, parentID, targetPath)
+}
+
+func (e *recoveringEngine) Delete(id string) (err error) {
+	defer recoverPanic(e.log, "Delete", &err)
+
+	return e.Engine.Delete(id)
+}
+
+func (e *recoveringEngine) Restore(id string) (err error) {
+	defer recoverPanic(e.log, "Restore", &err)
+
+	return e.Engine.Restore(id)
+}
+
+func (e *recoveringEngine) Purge(id string, onlyDeleted bool) (err error) {
+	defer recoverPanic(e.log, "Purge", &err)
+
+	return e.Engine.Purge(id, onlyDeleted)
+}
+
+func (e *recoveringEngine) PurgeSpace(rootID string) (err error) {
+	defer recoverPanic(e.log, "PurgeSpace", &err)
+
+	return e.Engine.PurgeSpace(rootID)
+}
+
+// NewBatch doesn't recover, creating a batch only allocates and never touches
+// the index. It wraps the returned batch so its index writes are guarded too.
+func (e *recoveringEngine) NewBatch(batchSize int) (BatchOperator, error) {
+	next, err := e.Engine.NewBatch(batchSize)
+	if err != nil || next == nil {
+		return next, err
+	}
+
+	return &recoveringBatch{BatchOperator: next, log: e.log}, nil
+}
+
+// recoveringBatch guards the batch calls that read from or write to the index.
+// Upsert is promoted, it only stages a document in the in-memory batch.
+type recoveringBatch struct {
+	BatchOperator
+
+	log log.Logger
+}
+
+func (b *recoveringBatch) Move(id string, parentID string, targetPath string) (err error) {
+	defer recoverPanic(b.log, "Batch.Move", &err)
+
+	return b.BatchOperator.Move(id, parentID, targetPath)
+}
+
+func (b *recoveringBatch) Delete(id string) (err error) {
+	defer recoverPanic(b.log, "Batch.Delete", &err)
+
+	return b.BatchOperator.Delete(id)
+}
+
+func (b *recoveringBatch) Restore(id string) (err error) {
+	defer recoverPanic(b.log, "Batch.Restore", &err)
+
+	return b.BatchOperator.Restore(id)
+}
+
+func (b *recoveringBatch) Purge(id string, onlyDeleted bool) (err error) {
+	defer recoverPanic(b.log, "Batch.Purge", &err)
+
+	return b.BatchOperator.Purge(id, onlyDeleted)
+}
+
+func (b *recoveringBatch) Push() (err error) {
+	defer recoverPanic(b.log, "Batch.Push", &err)
+
+	return b.BatchOperator.Push()
+}

services/search/pkg/search/recovering_test.go

@@ -0,0 +1,160 @@
+package search_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search/mocks"
+)
+
+func TestRecoveringEngineSearchRecoversPanic(t *testing.T) {
+	engine := mocks.NewEngine(t)
+	engine.EXPECT().Search(mock.Anything, mock.Anything).Run(func(context.Context, *searchService.SearchIndexRequest) {
+		panic("assignment to entry in nil map")
+	})
+
+	res, err := search.NewRecoveringEngine(engine, log.NopLogger()).Search(context.Background(), &searchService.SearchIndexRequest{})
+	if res != nil {
+		t.Errorf("expected a nil result, got %v", res)
+	}
+	if err == nil {
+		t.Fatal("expected an error, got nil")
+	}
+	if !strings.Contains(err.Error(), "search engine panic") {
+		t.Errorf("expected the error to name the engine panic, got %q", err.Error())
+	}
+	if !strings.Contains(err.Error(), "assignment to entry in nil map") {
+		t.Errorf("expected the error to carry the panic text, got %q", err.Error())
+	}
+}
+
+func TestRecoveringEngineSearchLogsPanic(t *testing.T) {
+	previousLevel := zerolog.GlobalLevel()
+	zerolog.SetGlobalLevel(zerolog.TraceLevel)
+	t.Cleanup(func() { zerolog.SetGlobalLevel(previousLevel) })
+
+	var buf bytes.Buffer
+	logger := log.Logger{Logger: zerolog.New(&buf)}
+
+	engine := mocks.NewEngine(t)
+	engine.EXPECT().Search(mock.Anything, mock.Anything).Run(func(context.Context, *searchService.SearchIndexRequest) {
+		panic("assignment to entry in nil map")
+	})
+
+	if _, err := search.NewRecoveringEngine(engine, logger).Search(context.Background(), &searchService.SearchIndexRequest{}); err == nil {
+		t.Fatal("expected an error, got nil")
+	}
+
+	out := buf.String()
+	if !strings.Contains(out, `"op":"Search"`) {
+		t.Errorf("expected the op field, got %q", out)
+	}
+	if !strings.Contains(out, `"stack":"`) || strings.Contains(out, `"stack":""`) {
+		t.Errorf("expected a non-empty stack field, got %q", out)
+	}
+}
+
+func TestRecoveringEngineUpsertRecoversPanic(t *testing.T) {
+	engine := mocks.NewEngine(t)
+	engine.EXPECT().Upsert(mock.Anything, mock.Anything).Run(func(string, search.Resource) {
+		panic("assignment to entry in nil map")
+	})
+
+	err := search.NewRecoveringEngine(engine, log.NopLogger()).Upsert("id", search.Resource{})
+	if err == nil {
+		t.Fatal("expected an error, got nil")
+	}
+	if !strings.Contains(err.Error(), "search engine panic") {
+		t.Errorf("expected the error to name the engine panic, got %q", err.Error())
+	}
+}
+
+func TestRecoveringEnginePassThrough(t *testing.T) {
+	sentinel := errors.New("index closed")
+
+	engine := mocks.NewEngine(t)
+	engine.EXPECT().DocCount().Return(uint64(42), nil).Once()
+	engine.EXPECT().DocCount().Return(uint64(0), sentinel).Once()
+
+	recovering := search.NewRecoveringEngine(engine, log.NopLogger())
+
+	count, err := recovering.DocCount()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if count != 42 {
+		t.Errorf("expected 42, got %d", count)
+	}
+
+	if _, err := recovering.DocCount(); !errors.Is(err, sentinel) {
+		t.Errorf("expected %v, got %v", sentinel, err)
+	}
+}
+
+func TestRecoveringBatchPushRecoversPanic(t *testing.T) {
+	batch := mocks.NewBatchOperator(t)
+	batch.EXPECT().Push().Run(func() {
+		panic("assignment to entry in nil map")
+	})
+
+	engine := mocks.NewEngine(t)
+	engine.EXPECT().NewBatch(mock.Anything).Return(batch, nil)
+
+	got, err := search.NewRecoveringEngine(engine, log.NopLogger()).NewBatch(10)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	err = got.Push()
+	if err == nil {
+		t.Fatal("expected an error, got nil")
+	}
+	if !strings.Contains(err.Error(), "search engine panic") {
+		t.Errorf("expected the error to name the engine panic, got %q", err.Error())
+	}
+}
+
+func TestRecoveringBatchPushPassThrough(t *testing.T) {
+	sentinel := errors.New("index closed")
+
+	batch := mocks.NewBatchOperator(t)
+	batch.EXPECT().Push().Return(sentinel)
+
+	engine := mocks.NewEngine(t)
+	engine.EXPECT().NewBatch(mock.Anything).Return(batch, nil)
+
+	got, err := search.NewRecoveringEngine(engine, log.NopLogger()).NewBatch(10)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	if err := got.Push(); !errors.Is(err, sentinel) {
+		t.Errorf("expected %v, got %v", sentinel, err)
+	}
+}
+
+func TestRecoveringBatchUpsertPassThrough(t *testing.T) {
+	batch := mocks.NewBatchOperator(t)
+	batch.EXPECT().Upsert(mock.Anything, mock.Anything).Return(nil)
+
+	engine := mocks.NewEngine(t)
+	engine.EXPECT().NewBatch(mock.Anything).Return(batch, nil)
+
+	got, err := search.NewRecoveringEngine(engine, log.NopLogger()).NewBatch(10)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	if err := got.Upsert("id", search.Resource{}); err != nil {
+		t.Errorf("expected no error, got %v", err)
+	}
+}