diff --git a/internal/application/repository/retriever/qdrant/repository.go b/internal/application/repository/retriever/qdrant/repository.go index 5990baa50..e38cd5ce5 100644 --- a/internal/application/repository/retriever/qdrant/repository.go +++ b/internal/application/repository/retriever/qdrant/repository.go @@ -165,7 +165,15 @@ func (q *qdrantRepository) ensureCollection(ctx context.Context, dimension int) ShardNumber: types.OptionalUint32(q.shardNumber), ReplicationFactor: types.OptionalUint32(q.replicationFactor), }) - if err != nil { + // The first batches written at a new dimension are saved by several + // workers at once, so another one (or another replica) may have + // created the collection since our check. Qdrant refuses the losing + // create with AlreadyExists; carry on, since the collection is there. + // Creating a payload index that already exists is a no-op, so the + // indexes below are still ensured in case the winner did not finish. + if status.Code(err) == codes.AlreadyExists { + log.Infof("[Qdrant] Collection %s was created concurrently", collectionName) + } else if err != nil { log.Errorf("[Qdrant] Failed to create collection: %v", err) return fmt.Errorf("failed to create collection: %w", err) } @@ -213,7 +221,7 @@ func (q *qdrantRepository) ensureCollection(ctx context.Context, dimension int) log.Warnf("[Qdrant] Failed to create text index for content: %v", err) } - log.Infof("[Qdrant] Successfully created collection %s", collectionName) + log.Infof("[Qdrant] Collection %s is ready", collectionName) } // Mark as initialized diff --git a/internal/application/repository/retriever/qdrant/repository_test.go b/internal/application/repository/retriever/qdrant/repository_test.go index 87c5bf54d..2f2666e2c 100644 --- a/internal/application/repository/retriever/qdrant/repository_test.go +++ b/internal/application/repository/retriever/qdrant/repository_test.go @@ -881,3 +881,62 @@ func TestPointWritesWaitForApplication(t *testing.T) { }) } } + +// newEnsureCollectionTestRepository answers the existence probe with "missing" +// and CreateCollection with createErr, recording the payload indexes created. +func newEnsureCollectionTestRepository(t *testing.T, createErr error) (*qdrantRepository, *[]string) { + t.Helper() + var indexed []string + client := newInterceptedQdrantClient(t, func( + _ context.Context, method string, req, reply any, _ *grpc.ClientConn, + _ grpc.UnaryInvoker, _ ...grpc.CallOption, + ) error { + switch request := req.(type) { + case *qdrant.CollectionExistsRequest: + reply.(*qdrant.CollectionExistsResponse).Result = &qdrant.CollectionExists{Exists: false} + return nil + case *qdrant.CreateCollection: + return createErr + case *qdrant.CreateFieldIndexCollection: + indexed = append(indexed, request.GetFieldName()) + return nil + default: + return fmt.Errorf("unexpected RPC %s", method) + } + }) + return &qdrantRepository{client: client, collectionBaseName: "vectors"}, &indexed +} + +// Several workers write the first batches of a new dimension at once, so the +// create can lose to another worker. That must not fail the write: the +// collection exists, and its payload indexes are still ensured. +func TestEnsureCollectionAcceptsConcurrentCreate(t *testing.T) { + repo, indexed := newEnsureCollectionTestRepository(t, + status.Error(codes.AlreadyExists, "Wrong input: Collection `vectors_1024` already exists!")) + + if err := repo.ensureCollection(context.Background(), 1024); err != nil { + t.Fatalf("ensureCollection after losing the create race: %v", err) + } + if _, ok := repo.initializedCollections.Load(1024); !ok { + t.Fatal("dimension must be cached once the collection is known to exist") + } + want := []string{fieldChunkID, fieldKnowledgeID, fieldKnowledgeBaseID, fieldSourceID, fieldIsEnabled, fieldContent} + if !slices.Equal(*indexed, want) { + t.Fatalf("payload indexes = %v, want %v", *indexed, want) + } +} + +func TestEnsureCollectionReturnsOtherCreateErrors(t *testing.T) { + repo, indexed := newEnsureCollectionTestRepository(t, status.Error(codes.Unavailable, "qdrant down")) + + err := repo.ensureCollection(context.Background(), 1024) + if status.Code(err) != codes.Unavailable { + t.Fatalf("ensureCollection error = %v, want the Unavailable create error", err) + } + if _, ok := repo.initializedCollections.Load(1024); ok { + t.Fatal("a failed create must not cache the dimension") + } + if len(*indexed) != 0 { + t.Fatalf("no payload index should be created after a failed create, got %v", *indexed) + } +}