diff --git a/weed/s3api/s3tables/vector_namespace.go b/weed/s3api/s3tables/vector_namespace.go index 028518683b..f07ffa9794 100644 --- a/weed/s3api/s3tables/vector_namespace.go +++ b/weed/s3api/s3tables/vector_namespace.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" ) // VectorIndexNamespace holds a vector bucket's indexes. Only the S3 Vectors @@ -53,9 +54,9 @@ func (h *S3TablesHandler) checkVectorNamespaceWrite(ctx context.Context, filerCl // fenceVectorNamespace returns a fence for an untrusted write naming the // vector namespace of a LANCE bucket. The request still runs, so the -// operation's own checks answer everything short of a write and an ungranted -// caller gets its usual denial. A CreateTable of an index that already exists -// is the vector engine's declare and passes unfenced. +// operation's own checks answer everything short of a write: an ungranted +// caller gets its usual denial, and a CreateTable that finds the index already +// there (the vector engine's declare) succeeds. func (h *S3TablesHandler) fenceVectorNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient, operation string) *vectorNamespaceFence { if h.trusted || strings.HasPrefix(operation, "Get") || strings.HasPrefix(operation, "List") { return nil @@ -69,7 +70,6 @@ func (h *S3TablesHandler) fenceVectorNamespace(w http.ResponseWriter, r *http.Re SourceNamespace []string `json:"sourceNamespace"` DestNamespace []string `json:"destNamespace"` ResourceARN string `json:"resourceArn"` - Name string `json:"name"` } // Like the handler's own decode, this keeps going past a mistyped field. _ = json.Unmarshal(body, &req) @@ -83,34 +83,13 @@ func (h *S3TablesHandler) fenceVectorNamespace(w http.ResponseWriter, r *http.Re targets = append(targets, target{bucket, []string{namespace}}) } for _, t := range targets { - if t.bucket == "" || h.checkVectorNamespaceWrite(r.Context(), filerClient, t.bucket, t.namespace) == nil { - continue + if t.bucket != "" && h.checkVectorNamespaceWrite(r.Context(), filerClient, t.bucket, t.namespace) != nil { + return &vectorNamespaceFence{ResponseWriter: w, h: h, filer: filerClient} } - if operation == "CreateTable" && isVectorIndexNamespace(req.Namespace) && h.vectorIndexExists(r.Context(), filerClient, bucketName, req.Name) { - return nil - } - return &vectorNamespaceFence{ResponseWriter: w, h: h, filer: filerClient} } return nil } -func (h *S3TablesHandler) vectorIndexExists(ctx context.Context, filerClient FilerClient, bucketName, name string) bool { - if _, err := validateTableName(name); err != nil { - return false - } - exists := false - _ = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - entry, err := h.lookupEntry(ctx, client, GetTablePath(bucketName, VectorIndexNamespace, name)) - if err != nil { - return err - } - _, hasMetadata := entry.Extended[ExtendedKeyMetadata] - exists = hasMetadata && !isUnfinishedCreate(entry.Extended) && EntryType(entry.Extended) == EntryTypeTable - return nil - }) - return exists -} - // vectorNamespaceFence passes reads, refuses filer writes, and answers the // request as read-only once a write was refused. Handlers write only after // their permission check, so an ungranted caller never sees that answer. @@ -169,7 +148,16 @@ func (c readOnlyFilerClient) CreateEntry(context.Context, *filer_pb.CreateEntryR return nil, c.refuse() } -func (c readOnlyFilerClient) UpdateEntry(context.Context, *filer_pb.UpdateEntryRequest, ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { +// UpdateEntry answers a rewrite that leaves the entry exactly as stored +// without writing, so an idempotent path that repairs state already in place +// still passes. +func (c readOnlyFilerClient) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + if req.GetEntry() != nil { + resp, err := c.SeaweedFilerClient.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{Directory: req.Directory, Name: req.Entry.Name}, opts...) + if err == nil && proto.Equal(resp.GetEntry(), req.Entry) { + return &filer_pb.UpdateEntryResponse{}, nil + } + } return nil, c.refuse() } diff --git a/weed/s3api/s3tables/vector_namespace_test.go b/weed/s3api/s3tables/vector_namespace_test.go index 4f43aada37..46bd0e6102 100644 --- a/weed/s3api/s3tables/vector_namespace_test.go +++ b/weed/s3api/s3tables/vector_namespace_test.go @@ -5,6 +5,7 @@ import ( "strings" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" "github.com/stretchr/testify/assert" @@ -264,24 +265,26 @@ func TestAuthorizeCreateTableRefusesVectorNamespace(t *testing.T) { assert.NoError(t, untrustedManager().AuthorizeCreateTable(ctx, NewManagerClient(icebergFS.Client), req, "reader")) } -// Only a CreateTable that finds a finished index passes unfenced, so the -// declare keeps working even where the idempotent path writes. -func TestVectorNamespaceFenceSparesOnlyExistingIndexCreate(t *testing.T) { +// Behind the fence a rewrite that leaves the entry as stored, as an +// idempotent repair does, passes without writing; any real change is refused. +func TestVectorNamespaceFencePassesOnlyNoOpUpdates(t *testing.T) { fs := s3tablestest.Start(t) seedVectorBucket(t, fs, FormatLance) - fs.Put(GetNamespacePath(vectorTestBucket, VectorIndexNamespace), "pending", map[string][]byte{ - ExtendedKeyMetadata: []byte(`{"name":"pending","format":"LANCE"}`), - ExtendedKeyDeleting: []byte(createClaim), - }) h := untrustedManager().handler + fence := &vectorNamespaceFence{h: h, filer: NewManagerClient(fs.Client)} + idxPath := GetTablePath(vectorTestBucket, VectorIndexNamespace, "idx") + stored := fs.Get(GetNamespacePath(vectorTestBucket, VectorIndexNamespace), "idx").Extended[ExtendedKeyMetadata] - for name, fenced := range map[string]bool{"idx": false, "fresh": true, "pending": true} { - t.Run(name, func(t *testing.T) { - body := `{"tableBucketARN":"` + vectorTestARN(t) + `","namespace":["vector"],"name":"` + name + `","format":"LANCE"}` - req, err := newManagerRequest(context.Background(), "CreateTable", []byte(body), "reader") - require.NoError(t, err) - got := h.fenceVectorNamespace(nil, req, NewManagerClient(fs.Client), "CreateTable") - assert.Equal(t, fenced, got != nil) - }) - } + err := fence.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return h.setExtendedAttribute(context.Background(), client, idxPath, ExtendedKeyMetadata, stored) + }) + require.NoError(t, err) + assert.False(t, fence.tripped.Load()) + + err = fence.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return h.setExtendedAttribute(context.Background(), client, idxPath, ExtendedKeyMetadata, []byte(`{}`)) + }) + assert.ErrorIs(t, err, errVectorNamespaceReadOnly) + assert.True(t, fence.tripped.Load()) + assert.Equal(t, stored, fs.Get(GetNamespacePath(vectorTestBucket, VectorIndexNamespace), "idx").Extended[ExtendedKeyMetadata]) }