diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index f0bf1d6371..99ee449afd 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -527,6 +527,11 @@ message DeleteEntryRequest { int64 if_not_modified_after = 9; // unix time in seconds bool keep_remote_object = 10; int64 if_not_modified_after_ns = 11; // unix time in nanoseconds; overrides if_not_modified_after when set + // Optional precondition evaluated against the entry atomically with the + // delete, under the filer's per-path lock. The request is routed to the + // entry's owner filer for the check to be authoritative. + map expected_extended = 12; + bool is_moved = 13; // set on a forwarded request so the owner applies it locally } message DeleteEntryResponse { diff --git a/seaweed-vector/crates/core/src/query.rs b/seaweed-vector/crates/core/src/query.rs index 420afb69fb..89c33f3f0f 100644 --- a/seaweed-vector/crates/core/src/query.rs +++ b/seaweed-vector/crates/core/src/query.rs @@ -167,6 +167,9 @@ mod tests { use super::*; use crate::table::VectorTable; use crate::testing::{DIM, write_dataset}; + use arrow_array::{FixedSizeListArray, RecordBatchIterator, types::Float32Type}; + use arrow_schema::{DataType, Field, Schema}; + use std::sync::Arc; use tempfile::TempDir; fn l2(a: &[f32], b: &[f32]) -> f32 { @@ -240,6 +243,135 @@ mod tests { assert_eq!(hits[1].key, "k2"); } + #[tokio::test] + async fn filter_matches_json_types_exactly() { + let dir = TempDir::new().unwrap(); + let metas = vec![ + r#"{"k": 3}"#, // k0: number + r#"{"k": "3"}"#, // k1: numeric string + r#"{"k": true}"#, // k2: bool + r#"{"k": "abc"}"#, // k3: non-numeric string + r#"{"x": 1}"#, // k4: missing key + r#"{"k": [1, 2]}"#, // k5: array + r#"{"k": null}"#, // k6: null + r#"{"k": 1}"#, // k7: serializes to '1', castable to boolean + r#"{"k": 0}"#, // k8: serializes to '0', castable to boolean + ]; + let schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Utf8, false), + Field::new( + "vector", + DataType::FixedSizeList( + Arc::new(Field::new("item", DataType::Float32, true)), + DIM as i32, + ), + false, + ), + Field::new("metadata", DataType::Utf8, true).with_metadata( + [("ARROW:extension:name".to_string(), "arrow.json".to_string())] + .into_iter() + .collect(), + ), + ])); + let keys: Vec = (0..metas.len()).map(|i| format!("k{i}")).collect(); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(keys)), + Arc::new( + FixedSizeListArray::from_iter_primitive::( + (0..metas.len()) + .map(|i| Some(vec![Some(i as f32); DIM])) + .collect::>(), + DIM as i32, + ), + ), + Arc::new(StringArray::from( + metas + .iter() + .map(|m| Some(m.to_string())) + .collect::>(), + )), + ], + ) + .unwrap(); + let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema); + let dataset = Dataset::write(reader, dir.path().to_str().unwrap(), None) + .await + .unwrap(); + let table = VectorTable::resolve(&dataset, None, None).unwrap(); + async fn run(dataset: &Dataset, table: &VectorTable, filter: &str) -> Vec { + flat( + dataset, + table, + &Query { + vector: vec![1.0, 0.0, 0.0, 0.0], + k: 10, + filter: Some(filter.to_string()), + postfilter: false, + distance: DistanceType::L2, + nprobes: None, + }, + ) + .await + .unwrap() + } + let keys_of = |hits: Vec| { + let mut keys: Vec = hits.into_iter().map(|h| h.key).collect(); + keys.sort(); + keys + }; + + // A stored string, boolean, array, or null must not satisfy a numeric + // comparison — nor fail the query, the way the converting json_get_* + // extractors do. + let hits = run( + &dataset, + &table, + "TRY_CAST(json_extract(`metadata`, '$[\"k\"]') AS DOUBLE) = 3", + ) + .await; + assert_eq!(keys_of(hits), ["k0"]); + let hits = run( + &dataset, + &table, + "TRY_CAST(json_extract(`metadata`, '$[\"k\"]') AS DOUBLE) > 1", + ) + .await; + assert_eq!(keys_of(hits), ["k0"]); + + // Booleans compare the serialized form as well: '1' and '0' would + // satisfy a cast but not the text match. + let hits = run( + &dataset, + &table, + "json_extract(`metadata`, '$[\"k\"]') = 'true'", + ) + .await; + assert_eq!(keys_of(hits), ["k2"]); + + // The serialized extract is exact for strings too: the numeric 3 is + // not a match for '"3"', and comparing json_extract never trips on + // the array or null rows. + let hits = run( + &dataset, + &table, + "json_extract(`metadata`, '$[\"k\"]') = '\"3\"'", + ) + .await; + assert_eq!(keys_of(hits), ["k1"]); + let hits = run( + &dataset, + &table, + "NOT COALESCE(json_extract(`metadata`, '$[\"k\"]') = '\"3\"', false)", + ) + .await; + assert_eq!( + keys_of(hits), + ["k0", "k2", "k3", "k4", "k5", "k6", "k7", "k8"] + ); + } + #[tokio::test] async fn get_fetches_rows_by_key() { let dir = TempDir::new().unwrap(); diff --git a/weed/command/s3.go b/weed/command/s3.go index 1cdf887ab9..2d3a0137ad 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -5,7 +5,7 @@ import ( "crypto/tls" "crypto/x509" "fmt" - "io/ioutil" + "net" "net/http" "os" "runtime" @@ -507,14 +507,9 @@ func (s3opt *S3Options) startS3Server() bool { } grace.OnInterrupt(certProvider.Close) - caCertPool := x509.NewCertPool() - if *s3opt.tlsCACertificate != "" { - // load CA certificate file and add it to list of client CAs - caCertFile, err := ioutil.ReadFile(*s3opt.tlsCACertificate) - if err != nil { - glog.Fatalf("error reading CA certificate: %v", err) - } - caCertPool.AppendCertsFromPEM(caCertFile) + caCertPool, err := loadClientCAPool(*s3opt.tlsCACertificate) + if err != nil { + glog.Fatalf("%v", err) } clientAuth := tls.NoClientCert @@ -726,6 +721,48 @@ func (s3opt *S3Options) startVectorsServer(s3ApiServer *s3api.S3ApiServer, grpcD vectorsServer.SetAccountID(s3tables.DefaultAccountID) vectorsServer.RegisterRoutes(vectorsRouter) + // Mirror the main API: when TLS is configured the vectors port serves + // HTTPS, not a second plaintext endpoint. Build the TLS config before + // binding listeners so a certificate failure does not leave ports bound. + var tlsConfig *tls.Config + if *s3opt.tlsPrivateKey != "" { + // The reloading provider reads the key pair asynchronously, so validate + // it once up front: a missing or mismatched pair must fail here rather + // than after the listeners bind an endpoint that cannot handshake. + if _, err := tls.LoadX509KeyPair(*s3opt.tlsCertificate, *s3opt.tlsPrivateKey); err != nil { + glog.Errorf("S3 Vectors: load HTTPS certificate: %v", err) + conn.Close() + return nil + } + getCert, _, err := security.NewReloadingServerCertificate(*s3opt.tlsCertificate, *s3opt.tlsPrivateKey) + if err != nil { + glog.Errorf("S3 Vectors: load HTTPS certificate: %v", err) + conn.Close() + return nil + } + clientAuth := tls.NoClientCert + var caCertPool *x509.CertPool + if *s3opt.tlsVerifyClientCert { + clientAuth = tls.RequireAndVerifyClientCert + caCertPool, err = loadClientCAPool(*s3opt.tlsCACertificate) + if err != nil { + glog.Errorf("S3 Vectors: %v", err) + conn.Close() + return nil + } + } + tlsConfig = &tls.Config{ + GetCertificate: getCert, + ClientAuth: clientAuth, + ClientCAs: caCertPool, + } + if err := security.FixTlsConfig(util.GetViper(), tlsConfig); err != nil { + glog.Errorf("S3 Vectors: tls config: %v", err) + conn.Close() + return nil + } + } + listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.portVectors) vectorsListener, vectorsLocalListener, err := util.NewIpAndLocalListeners( *s3opt.bindIp, *s3opt.portVectors, time.Duration(*s3opt.idleTimeout)*time.Second, listenOnLocalhost) @@ -735,57 +772,51 @@ func (s3opt *S3Options) startVectorsServer(s3ApiServer *s3api.S3ApiServer, grpcD return nil } - // Mirror the main API: when TLS is configured the vectors port serves - // HTTPS, not a second plaintext endpoint. - if *s3opt.tlsPrivateKey != "" { - getCert, _, err := security.NewReloadingServerCertificate(*s3opt.tlsCertificate, *s3opt.tlsPrivateKey) - if err != nil { - glog.Errorf("S3 Vectors: load HTTPS certificate: %v", err) - return nil - } - clientAuth := tls.NoClientCert - if *s3opt.tlsVerifyClientCert { - clientAuth = tls.RequireAndVerifyClientCert - } - tlsConfig := &tls.Config{GetCertificate: getCert, ClientAuth: clientAuth} - if err := security.FixTlsConfig(util.GetViper(), tlsConfig); err != nil { - glog.Errorf("S3 Vectors: tls config: %v", err) - return nil - } - httpS := newHttpServer(vectorsRouter, tlsConfig) - if vectorsLocalListener != nil { - go func() { - if err := httpS.ServeTLS(vectorsLocalListener, "", ""); err != nil && err != http.ErrServerClosed { - glog.V(0).Infof("S3 Vectors localhost listener error: %v", err) - } - }() - } - go func() { - if err := httpS.ServeTLS(vectorsListener, "", ""); err != nil && err != http.ErrServerClosed { - glog.Errorf("S3 Vectors API Server Fail to serve: %v", err) - } - }() - return httpS + if tlsConfig != nil { + glog.V(0).Infof("Start S3 Vectors API Server at https://%s", listenAddress) + } else { + glog.V(0).Infof("Start S3 Vectors API Server at http://%s", listenAddress) } - glog.V(0).Infof("Start S3 Vectors API Server at http://%s", listenAddress) - - httpS := newHttpServer(vectorsRouter, nil) + httpS := newHttpServer(vectorsRouter, tlsConfig) + serve := func(listener net.Listener) error { + if tlsConfig != nil { + return httpS.ServeTLS(listener, "", "") + } + return httpS.Serve(listener) + } if vectorsLocalListener != nil { go func() { - if err := httpS.Serve(vectorsLocalListener); err != nil && err != http.ErrServerClosed { + if err := serve(vectorsLocalListener); err != nil && err != http.ErrServerClosed { glog.V(0).Infof("S3 Vectors localhost listener error: %v", err) } }() } go func() { - if err := httpS.Serve(vectorsListener); err != nil && err != http.ErrServerClosed { + if err := serve(vectorsListener); err != nil && err != http.ErrServerClosed { glog.Errorf("S3 Vectors API Server Fail to serve: %v", err) } }() return httpS } +// loadClientCAPool builds the client CA pool from -cacert.file, failing when +// the file reads but carries no usable certificate. +func loadClientCAPool(caCertFile string) (*x509.CertPool, error) { + pool := x509.NewCertPool() + if caCertFile == "" { + return pool, nil + } + pem, err := os.ReadFile(caCertFile) + if err != nil { + return nil, fmt.Errorf("error reading CA certificate: %w", err) + } + if !pool.AppendCertsFromPEM(pem) { + return nil, fmt.Errorf("no valid certificates found in %s", caCertFile) + } + return pool, nil +} + // deriveLanceStorageEndpoint picks the endpoint the Lance namespace puts in // storage_options. It falls back to the advertised -ip where the Iceberg // derivation gives up, because the two clients are not in the same position: a diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 1fde04dfd2..b4024c9291 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -527,6 +527,11 @@ message DeleteEntryRequest { int64 if_not_modified_after = 9; // unix time in seconds bool keep_remote_object = 10; int64 if_not_modified_after_ns = 11; // unix time in nanoseconds; overrides if_not_modified_after when set + // Optional precondition evaluated against the entry atomically with the + // delete, under the filer's per-path lock. The request is routed to the + // entry's owner filer for the check to be authoritative. + map expected_extended = 12; + bool is_moved = 13; // set on a forwarded request so the owner applies it locally } message DeleteEntryResponse { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 0cdfbadd48..163d9507eb 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v3.21.12 +// protoc v7.35.0 // source: filer.proto package filer_pb @@ -2783,8 +2783,13 @@ type DeleteEntryRequest struct { IfNotModifiedAfter int64 `protobuf:"varint,9,opt,name=if_not_modified_after,json=ifNotModifiedAfter,proto3" json:"if_not_modified_after,omitempty"` // unix time in seconds KeepRemoteObject bool `protobuf:"varint,10,opt,name=keep_remote_object,json=keepRemoteObject,proto3" json:"keep_remote_object,omitempty"` IfNotModifiedAfterNs int64 `protobuf:"varint,11,opt,name=if_not_modified_after_ns,json=ifNotModifiedAfterNs,proto3" json:"if_not_modified_after_ns,omitempty"` // unix time in nanoseconds; overrides if_not_modified_after when set - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Optional precondition evaluated against the entry atomically with the + // delete, under the filer's per-path lock. The request is routed to the + // entry's owner filer for the check to be authoritative. + ExpectedExtended map[string][]byte `protobuf:"bytes,12,rep,name=expected_extended,json=expectedExtended,proto3" json:"expected_extended,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + IsMoved bool `protobuf:"varint,13,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` // set on a forwarded request so the owner applies it locally + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *DeleteEntryRequest) Reset() { @@ -2887,6 +2892,20 @@ func (x *DeleteEntryRequest) GetIfNotModifiedAfterNs() int64 { return 0 } +func (x *DeleteEntryRequest) GetExpectedExtended() map[string][]byte { + if x != nil { + return x.ExpectedExtended + } + return nil +} + +func (x *DeleteEntryRequest) GetIsMoved() bool { + if x != nil { + return x.IsMoved + } + return false +} + type DeleteEntryResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` @@ -7086,7 +7105,7 @@ type LocateBrokerResponse_Resource struct { func (x *LocateBrokerResponse_Resource) Reset() { *x = LocateBrokerResponse_Resource{} - mi := &file_filer_proto_msgTypes[100] + mi := &file_filer_proto_msgTypes[101] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -7098,7 +7117,7 @@ func (x *LocateBrokerResponse_Resource) String() string { func (*LocateBrokerResponse_Resource) ProtoMessage() {} func (x *LocateBrokerResponse_Resource) ProtoReflect() protoreflect.Message { - mi := &file_filer_proto_msgTypes[100] + mi := &file_filer_proto_msgTypes[101] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -7154,7 +7173,7 @@ type FilerConf_PathConf struct { func (x *FilerConf_PathConf) Reset() { *x = FilerConf_PathConf{} - mi := &file_filer_proto_msgTypes[101] + mi := &file_filer_proto_msgTypes[102] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -7166,7 +7185,7 @@ func (x *FilerConf_PathConf) String() string { func (*FilerConf_PathConf) ProtoMessage() {} func (x *FilerConf_PathConf) ProtoReflect() protoreflect.Message { - mi := &file_filer_proto_msgTypes[101] + mi := &file_filer_proto_msgTypes[102] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -7569,7 +7588,7 @@ const file_filer_proto_rawDesc = "" + "\n" + "entry_name\x18\x02 \x01(\tR\tentryName\x12+\n" + "\x06chunks\x18\x03 \x03(\v2\x13.filer_pb.FileChunkR\x06chunks\"\x17\n" + - "\x15AppendToEntryResponse\"\xb1\x03\n" + + "\x15AppendToEntryResponse\"\xf2\x04\n" + "\x12DeleteEntryRequest\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x12\n" + "\x04name\x18\x02 \x01(\tR\x04name\x12$\n" + @@ -7583,7 +7602,12 @@ const file_filer_proto_rawDesc = "" + "\x15if_not_modified_after\x18\t \x01(\x03R\x12ifNotModifiedAfter\x12,\n" + "\x12keep_remote_object\x18\n" + " \x01(\bR\x10keepRemoteObject\x126\n" + - "\x18if_not_modified_after_ns\x18\v \x01(\x03R\x14ifNotModifiedAfterNs\"w\n" + + "\x18if_not_modified_after_ns\x18\v \x01(\x03R\x14ifNotModifiedAfterNs\x12_\n" + + "\x11expected_extended\x18\f \x03(\v22.filer_pb.DeleteEntryRequest.ExpectedExtendedEntryR\x10expectedExtended\x12\x19\n" + + "\bis_moved\x18\r \x01(\bR\aisMoved\x1aC\n" + + "\x15ExpectedExtendedEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\fR\x05value:\x028\x01\"w\n" + "\x13DeleteEntryResponse\x12\x14\n" + "\x05error\x18\x01 \x01(\tR\x05error\x12J\n" + "\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\"\xed\x01\n" + @@ -8025,7 +8049,7 @@ func file_filer_proto_rawDescGZIP() []byte { } var file_filer_proto_enumTypes = make([]protoimpl.EnumInfo, 5) -var file_filer_proto_msgTypes = make([]protoimpl.MessageInfo, 104) +var file_filer_proto_msgTypes = make([]protoimpl.MessageInfo, 105) var file_filer_proto_goTypes = []any{ (SSEType)(0), // 0: filer_pb.SSEType (FilerError)(0), // 1: filer_pb.FilerError @@ -8131,11 +8155,12 @@ var file_filer_proto_goTypes = []any{ nil, // 101: filer_pb.ObjectMutation.SetExtendedEntry nil, // 102: filer_pb.Recompute.CopyExtendedEntry nil, // 103: filer_pb.UpdateEntryRequest.ExpectedExtendedEntry - nil, // 104: filer_pb.LookupVolumeResponse.LocationsMapEntry - (*LocateBrokerResponse_Resource)(nil), // 105: filer_pb.LocateBrokerResponse.Resource - (*FilerConf_PathConf)(nil), // 106: filer_pb.FilerConf.PathConf - nil, // 107: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry - nil, // 108: filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry + nil, // 104: filer_pb.DeleteEntryRequest.ExpectedExtendedEntry + nil, // 105: filer_pb.LookupVolumeResponse.LocationsMapEntry + (*LocateBrokerResponse_Resource)(nil), // 106: filer_pb.LocateBrokerResponse.Resource + (*FilerConf_PathConf)(nil), // 107: filer_pb.FilerConf.PathConf + nil, // 108: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry + nil, // 109: filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry } var file_filer_proto_depIdxs = []int32{ 10, // 0: filer_pb.LookupDirectoryEntryResponse.entry:type_name -> filer_pb.Entry @@ -8175,118 +8200,119 @@ var file_filer_proto_depIdxs = []int32{ 18, // 34: filer_pb.UpdateEntryRequest.condition:type_name -> filer_pb.WriteCondition 59, // 35: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse 13, // 36: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk - 59, // 37: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse - 12, // 38: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification - 45, // 39: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location - 45, // 40: filer_pb.AssignVolumeResponse.replicas:type_name -> filer_pb.Location - 1, // 41: filer_pb.AssignVolumeResponse.error_code:type_name -> filer_pb.FilerError - 45, // 42: filer_pb.Locations.locations:type_name -> filer_pb.Location - 104, // 43: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry - 47, // 44: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection - 12, // 45: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification - 59, // 46: filer_pb.SubscribeMetadataResponse.events:type_name -> filer_pb.SubscribeMetadataResponse - 63, // 47: filer_pb.SubscribeMetadataResponse.log_file_refs:type_name -> filer_pb.LogFileChunkRef - 62, // 48: filer_pb.ListMetadataSubscribersResponse.subscribers:type_name -> filer_pb.MetadataSubscriber - 13, // 49: filer_pb.LogFileChunkRef.chunks:type_name -> filer_pb.FileChunk - 10, // 50: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry - 105, // 51: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource - 106, // 52: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf - 10, // 53: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry - 59, // 54: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse - 84, // 55: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock - 17, // 56: filer_pb.StreamMutateEntryRequest.create_request:type_name -> filer_pb.CreateEntryRequest - 29, // 57: filer_pb.StreamMutateEntryRequest.update_request:type_name -> filer_pb.UpdateEntryRequest - 35, // 58: filer_pb.StreamMutateEntryRequest.delete_request:type_name -> filer_pb.DeleteEntryRequest - 39, // 59: filer_pb.StreamMutateEntryRequest.rename_request:type_name -> filer_pb.StreamRenameEntryRequest - 28, // 60: filer_pb.StreamMutateEntryResponse.create_response:type_name -> filer_pb.CreateEntryResponse - 30, // 61: filer_pb.StreamMutateEntryResponse.update_response:type_name -> filer_pb.UpdateEntryResponse - 36, // 62: filer_pb.StreamMutateEntryResponse.delete_response:type_name -> filer_pb.DeleteEntryResponse - 40, // 63: filer_pb.StreamMutateEntryResponse.rename_response:type_name -> filer_pb.StreamRenameEntryResponse - 95, // 64: filer_pb.MountListResponse.mounts:type_name -> filer_pb.MountInfo - 5, // 65: filer_pb.LookupDirectoryEntriesRequest.requests:type_name -> filer_pb.LookupDirectoryEntryRequest - 10, // 66: filer_pb.LookupDirectoryEntryResult.entry:type_name -> filer_pb.Entry - 97, // 67: filer_pb.LookupDirectoryEntriesResponse.results:type_name -> filer_pb.LookupDirectoryEntryResult - 107, // 68: filer_pb.LookupDirectoryEntriesResponse.locations_map:type_name -> filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry - 108, // 69: filer_pb.LookupDirectoryEntriesResponse.read_auth:type_name -> filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry - 3, // 70: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind - 10, // 71: filer_pb.WriteCondition.Clause.expected_entry:type_name -> filer_pb.Entry - 44, // 72: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations - 44, // 73: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations - 5, // 74: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest - 96, // 75: filer_pb.SeaweedFiler.LookupDirectoryEntries:input_type -> filer_pb.LookupDirectoryEntriesRequest - 7, // 76: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest - 17, // 77: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest - 29, // 78: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest - 31, // 79: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest - 33, // 80: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest - 35, // 81: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest - 21, // 82: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest - 26, // 83: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest - 24, // 84: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest - 37, // 85: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest - 39, // 86: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest - 89, // 87: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest - 41, // 88: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest - 43, // 89: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest - 48, // 90: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest - 50, // 91: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest - 52, // 92: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest - 54, // 93: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest - 56, // 94: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest - 64, // 95: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest - 58, // 96: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 58, // 97: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 60, // 98: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest - 71, // 99: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest - 73, // 100: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest - 76, // 101: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest - 78, // 102: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest - 80, // 103: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest - 82, // 104: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest - 85, // 105: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest - 87, // 106: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest - 91, // 107: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest - 93, // 108: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest - 6, // 109: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse - 98, // 110: filer_pb.SeaweedFiler.LookupDirectoryEntries:output_type -> filer_pb.LookupDirectoryEntriesResponse - 8, // 111: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse - 28, // 112: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse - 30, // 113: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse - 32, // 114: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse - 34, // 115: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse - 36, // 116: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse - 22, // 117: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse - 27, // 118: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse - 25, // 119: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse - 38, // 120: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse - 40, // 121: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse - 90, // 122: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse - 42, // 123: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse - 46, // 124: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse - 49, // 125: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse - 51, // 126: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse - 53, // 127: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse - 55, // 128: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse - 57, // 129: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse - 65, // 130: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse - 59, // 131: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 59, // 132: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 61, // 133: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse - 72, // 134: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse - 74, // 135: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse - 77, // 136: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse - 79, // 137: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse - 81, // 138: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse - 83, // 139: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse - 86, // 140: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse - 88, // 141: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse - 92, // 142: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse - 94, // 143: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse - 109, // [109:144] is the sub-list for method output_type - 74, // [74:109] is the sub-list for method input_type - 74, // [74:74] is the sub-list for extension type_name - 74, // [74:74] is the sub-list for extension extendee - 0, // [0:74] is the sub-list for field type_name + 104, // 37: filer_pb.DeleteEntryRequest.expected_extended:type_name -> filer_pb.DeleteEntryRequest.ExpectedExtendedEntry + 59, // 38: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse + 12, // 39: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification + 45, // 40: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location + 45, // 41: filer_pb.AssignVolumeResponse.replicas:type_name -> filer_pb.Location + 1, // 42: filer_pb.AssignVolumeResponse.error_code:type_name -> filer_pb.FilerError + 45, // 43: filer_pb.Locations.locations:type_name -> filer_pb.Location + 105, // 44: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry + 47, // 45: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection + 12, // 46: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification + 59, // 47: filer_pb.SubscribeMetadataResponse.events:type_name -> filer_pb.SubscribeMetadataResponse + 63, // 48: filer_pb.SubscribeMetadataResponse.log_file_refs:type_name -> filer_pb.LogFileChunkRef + 62, // 49: filer_pb.ListMetadataSubscribersResponse.subscribers:type_name -> filer_pb.MetadataSubscriber + 13, // 50: filer_pb.LogFileChunkRef.chunks:type_name -> filer_pb.FileChunk + 10, // 51: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry + 106, // 52: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource + 107, // 53: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf + 10, // 54: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry + 59, // 55: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse + 84, // 56: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock + 17, // 57: filer_pb.StreamMutateEntryRequest.create_request:type_name -> filer_pb.CreateEntryRequest + 29, // 58: filer_pb.StreamMutateEntryRequest.update_request:type_name -> filer_pb.UpdateEntryRequest + 35, // 59: filer_pb.StreamMutateEntryRequest.delete_request:type_name -> filer_pb.DeleteEntryRequest + 39, // 60: filer_pb.StreamMutateEntryRequest.rename_request:type_name -> filer_pb.StreamRenameEntryRequest + 28, // 61: filer_pb.StreamMutateEntryResponse.create_response:type_name -> filer_pb.CreateEntryResponse + 30, // 62: filer_pb.StreamMutateEntryResponse.update_response:type_name -> filer_pb.UpdateEntryResponse + 36, // 63: filer_pb.StreamMutateEntryResponse.delete_response:type_name -> filer_pb.DeleteEntryResponse + 40, // 64: filer_pb.StreamMutateEntryResponse.rename_response:type_name -> filer_pb.StreamRenameEntryResponse + 95, // 65: filer_pb.MountListResponse.mounts:type_name -> filer_pb.MountInfo + 5, // 66: filer_pb.LookupDirectoryEntriesRequest.requests:type_name -> filer_pb.LookupDirectoryEntryRequest + 10, // 67: filer_pb.LookupDirectoryEntryResult.entry:type_name -> filer_pb.Entry + 97, // 68: filer_pb.LookupDirectoryEntriesResponse.results:type_name -> filer_pb.LookupDirectoryEntryResult + 108, // 69: filer_pb.LookupDirectoryEntriesResponse.locations_map:type_name -> filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry + 109, // 70: filer_pb.LookupDirectoryEntriesResponse.read_auth:type_name -> filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry + 3, // 71: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind + 10, // 72: filer_pb.WriteCondition.Clause.expected_entry:type_name -> filer_pb.Entry + 44, // 73: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 44, // 74: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 5, // 75: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest + 96, // 76: filer_pb.SeaweedFiler.LookupDirectoryEntries:input_type -> filer_pb.LookupDirectoryEntriesRequest + 7, // 77: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest + 17, // 78: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest + 29, // 79: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest + 31, // 80: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest + 33, // 81: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest + 35, // 82: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest + 21, // 83: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest + 26, // 84: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest + 24, // 85: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest + 37, // 86: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest + 39, // 87: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest + 89, // 88: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest + 41, // 89: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest + 43, // 90: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest + 48, // 91: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest + 50, // 92: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest + 52, // 93: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest + 54, // 94: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest + 56, // 95: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest + 64, // 96: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest + 58, // 97: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 58, // 98: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 60, // 99: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest + 71, // 100: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest + 73, // 101: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest + 76, // 102: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest + 78, // 103: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest + 80, // 104: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest + 82, // 105: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest + 85, // 106: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest + 87, // 107: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest + 91, // 108: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest + 93, // 109: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest + 6, // 110: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse + 98, // 111: filer_pb.SeaweedFiler.LookupDirectoryEntries:output_type -> filer_pb.LookupDirectoryEntriesResponse + 8, // 112: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse + 28, // 113: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse + 30, // 114: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse + 32, // 115: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse + 34, // 116: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse + 36, // 117: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse + 22, // 118: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse + 27, // 119: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse + 25, // 120: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse + 38, // 121: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse + 40, // 122: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse + 90, // 123: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse + 42, // 124: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse + 46, // 125: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse + 49, // 126: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse + 51, // 127: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse + 53, // 128: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse + 55, // 129: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse + 57, // 130: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse + 65, // 131: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse + 59, // 132: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 59, // 133: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 61, // 134: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse + 72, // 135: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse + 74, // 136: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse + 77, // 137: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse + 79, // 138: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse + 81, // 139: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse + 83, // 140: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse + 86, // 141: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse + 88, // 142: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse + 92, // 143: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse + 94, // 144: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse + 110, // [110:145] is the sub-list for method output_type + 75, // [75:110] is the sub-list for method input_type + 75, // [75:75] is the sub-list for extension type_name + 75, // [75:75] is the sub-list for extension extendee + 0, // [0:75] is the sub-list for field type_name } func init() { file_filer_proto_init() } @@ -8307,14 +8333,14 @@ func file_filer_proto_init() { (*StreamMutateEntryResponse_DeleteResponse)(nil), (*StreamMutateEntryResponse_RenameResponse)(nil), } - file_filer_proto_msgTypes[101].OneofWrappers = []any{} + file_filer_proto_msgTypes[102].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_filer_proto_rawDesc), len(file_filer_proto_rawDesc)), NumEnums: 5, - NumMessages: 104, + NumMessages: 105, NumExtensions: 0, NumServices: 1, }, diff --git a/weed/pb/filer_pb/filer_grpc.pb.go b/weed/pb/filer_pb/filer_grpc.pb.go index 8a330ce286..6a2b494a39 100644 --- a/weed/pb/filer_pb/filer_grpc.pb.go +++ b/weed/pb/filer_pb/filer_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.2 -// - protoc v3.21.12 +// - protoc v7.35.0 // source: filer.proto package filer_pb diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 6621620bed..5c5becb7bb 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -2491,6 +2491,35 @@ func (m *DeleteEntryRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.IsMoved { + i-- + if m.IsMoved { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x68 + } + if len(m.ExpectedExtended) > 0 { + for k := range m.ExpectedExtended { + v := m.ExpectedExtended[k] + baseI := i + i -= len(v) + copy(dAtA[i:], v) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(v))) + i-- + dAtA[i] = 0x12 + i -= len(k) + copy(dAtA[i:], k) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(k))) + i-- + dAtA[i] = 0xa + i = protohelpers.EncodeVarint(dAtA, i, uint64(baseI-i)) + i-- + dAtA[i] = 0x62 + } + } if m.IfNotModifiedAfterNs != 0 { i = protohelpers.EncodeVarint(dAtA, i, uint64(m.IfNotModifiedAfterNs)) i-- @@ -7579,6 +7608,18 @@ func (m *DeleteEntryRequest) SizeVT() (n int) { if m.IfNotModifiedAfterNs != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.IfNotModifiedAfterNs)) } + if len(m.ExpectedExtended) > 0 { + for k, v := range m.ExpectedExtended { + _ = k + _ = v + l = 1 + len(v) + protohelpers.SizeOfVarint(uint64(len(v))) + mapEntrySize := 1 + len(k) + protohelpers.SizeOfVarint(uint64(len(k))) + l + n += mapEntrySize + 1 + protohelpers.SizeOfVarint(uint64(mapEntrySize)) + } + } + if m.IsMoved { + n += 2 + } n += len(m.unknownFields) return n } @@ -16232,6 +16273,154 @@ func (m *DeleteEntryRequest) UnmarshalVT(dAtA []byte) error { break } } + case 12: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ExpectedExtended", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.ExpectedExtended == nil { + m.ExpectedExtended = make(map[string][]byte) + } + var mapkey string + var mapvalue []byte + for iNdEx < postIndex { + entryPreIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + if fieldNum == 1 { + var stringLenmapkey uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapkey |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapkey := int(stringLenmapkey) + if intStringLenmapkey < 0 { + return protohelpers.ErrInvalidLength + } + postStringIndexmapkey := iNdEx + intStringLenmapkey + if postStringIndexmapkey < 0 { + return protohelpers.ErrInvalidLength + } + if postStringIndexmapkey > l { + return io.ErrUnexpectedEOF + } + mapkey = string(dAtA[iNdEx:postStringIndexmapkey]) + iNdEx = postStringIndexmapkey + } else if fieldNum == 2 { + var mapbyteLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + mapbyteLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intMapbyteLen := int(mapbyteLen) + if intMapbyteLen < 0 { + return protohelpers.ErrInvalidLength + } + postbytesIndex := iNdEx + intMapbyteLen + if postbytesIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postbytesIndex > l { + return io.ErrUnexpectedEOF + } + mapvalue = make([]byte, mapbyteLen) + copy(mapvalue, dAtA[iNdEx:postbytesIndex]) + iNdEx = postbytesIndex + } else { + iNdEx = entryPreIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > postIndex { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + m.ExpectedExtended[mapkey] = mapvalue + iNdEx = postIndex + case 13: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field IsMoved", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.IsMoved = bool(v != 0) default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 4ed73cbc39..aec770ccc9 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -17,6 +17,7 @@ import ( var ( ErrAttributeNotFound = errors.New("attribute not found") ErrConcurrentUpdate = errors.New("attribute changed concurrently") + ErrNameTaken = errors.New("catalog name is taken") ) // extendedAttributeUpdateAttempts bounds the read-modify-write retry loop in @@ -179,6 +180,75 @@ func (h *S3TablesHandler) removeExtendedAttributesIf( }) } +// claimCatalogEntry writes a new catalog entry's marker attributes in one +// compare-checked update. Every catalog marker must still be absent: a +// delete's restore or a competing create that landed since the caller's check +// is a name conflict, not a record this write may overwrite — an overwrite +// would leave the loser's tags, policy, and maintenance settings on the new +// record. +func (h *S3TablesHandler) claimCatalogEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, attrs map[string][]byte) error { + return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error { + for _, key := range s3tablesExtendedKeys { + if _, ok := extended[key]; ok { + return fmt.Errorf("%w: %s set on %s", ErrNameTaken, key, path) + } + } + for key, data := range attrs { + extended[key] = data + } + return nil + }) +} + +// claimCatalogRecord reserves a catalog record for an in-flight delete or +// rename under one compare-checked write: the entry must still hold exactly +// expected. A record claimed by another kind of operation refuses the claim, +// so the loser backs out before any data or destination entry is written; +// a retry of the same kind may adopt its own earlier claim. +func (h *S3TablesHandler) claimCatalogRecord( + ctx context.Context, + client filer_pb.SeaweedFilerClient, + path string, + expected map[string][]byte, + claim string, +) error { + return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error { + for key, want := range expected { + if key == ExtendedKeyDeleting { + continue + } + if !bytes.Equal(extended[key], want) { + return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, path) + } + } + if cur, ok := extended[ExtendedKeyDeleting]; ok && string(cur) != claim { + return fmt.Errorf("%w: %s held on %s", ErrConcurrentUpdate, ExtendedKeyDeleting, path) + } + extended[ExtendedKeyDeleting] = []byte(claim) + return nil + }) +} + +// releaseCatalogRecord lifts a claimCatalogRecord reservation while the entry +// still holds expected — which includes the claim itself — restoring the +// record so the failed operation can be retried. +func (h *S3TablesHandler) releaseCatalogRecord( + ctx context.Context, + client filer_pb.SeaweedFilerClient, + path string, + expected map[string][]byte, +) error { + return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error { + for key, want := range expected { + if !bytes.Equal(extended[key], want) { + return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, path) + } + } + delete(extended, ExtendedKeyDeleting) + return nil + }) +} + // updateExtendedAttribute applies mutate to the current value of key and writes // the result back. mutate runs again on each retry, so it always sees the // current value. @@ -211,6 +281,7 @@ var s3tablesExtendedKeys = []string{ ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyEntryType, + ExtendedKeyDeleting, } // SnapshotExtended captures an entry's attributes as an UpdateEntry @@ -314,6 +385,27 @@ func (h *S3TablesHandler) deleteDirectory(ctx context.Context, client filer_pb.S return filer_pb.DoRemove(ctx, client, dir, name, true, true, true, false, nil) } +// deleteDirectoryIfMatches deletes a directory and all its contents only +// while the entry still carries the extended attributes the caller captured: +// the filer compares and removes in one operation under the path lock, so an +// attribute write that keeps Mtime or a delete+recreate in between refuses the +// delete with ErrConcurrentUpdate instead of losing the replacement. +func (h *S3TablesHandler) deleteDirectoryIfMatches(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, expected *filer_pb.Entry) error { + dir, name := splitPath(path) + _, err := filer_pb.DoRemoveRequest(ctx, client, &filer_pb.DeleteEntryRequest{ + Directory: dir, + Name: name, + IsDeleteData: true, + IsRecursive: true, + IgnoreRecursiveError: true, + ExpectedExtended: snapshotExtended(expected.Extended), + }) + if status.Code(err) == codes.FailedPrecondition { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path) + } + return err +} + // entryExists checks if an entry exists at the given path func (h *S3TablesHandler) entryExists(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) bool { dir, name := splitPath(path) diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index 4bf960ca4f..eda5e9ae6c 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -32,6 +32,10 @@ const ( // contend on the same attribute. ExtendedKeyMaintenanceStatus = s3_constants.ExtS3TablesPrefix + "maintenanceStatus" ExtendedKeyEntryType = s3_constants.ExtS3TablesPrefix + "entryType" + // ExtendedKeyDeleting marks a catalog record claimed by an in-flight + // delete or rename, so the claim survives until the operation finishes + // or releases it. + ExtendedKeyDeleting = s3_constants.ExtS3TablesPrefix + "deleting" // Entry-type marker values for ExtendedKeyEntryType. Absent or "table" means // a table; views are stored like tables but tagged "view". diff --git a/weed/s3api/s3tables/handler_delete_decouple_test.go b/weed/s3api/s3tables/handler_delete_decouple_test.go index f574caf63f..c22c433325 100644 --- a/weed/s3api/s3tables/handler_delete_decouple_test.go +++ b/weed/s3api/s3tables/handler_delete_decouple_test.go @@ -3,6 +3,7 @@ package s3tables import ( "context" "encoding/json" + "net/http" "path" "testing" @@ -21,6 +22,86 @@ func runDeleteTable(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, namespa }, nil, "") } +// A write that lands between the delete's read and its conditional commit — +// an xattr update keeps mtime — must turn a token delete into a conflict, not +// get erased by it. +func TestDeleteTableTokenDeleteLosesToConcurrentUpdate(t *testing.T) { + fs, m := startRenameManager(t) + + token := "tok-ct" + meta, _ := json.Marshal(tableMetadataInternal{ + Name: "ct", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + VersionToken: token, + }) + fs.Put(GetNamespacePath(renameTestBucket, "ns"), "ct", map[string][]byte{ + ExtendedKeyMetadata: meta, + }) + tablePath := GetTablePath(renameTestBucket, "ns", "ct") + fs.Put(tablePath, "data", nil) + + fs.BeforeDelete = func() { + fs.Put(GetNamespacePath(renameTestBucket, "ns"), "ct", map[string][]byte{ + ExtendedKeyMetadata: meta, + ExtendedKeyTags: []byte(`{"k":"v"}`), + }) + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTable", &DeleteTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "ct", + VersionToken: token, + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeConflict, resp.Type) + + assert.NotNil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "ct"), "the updated record must survive") + assert.NotNil(t, fs.Get(tablePath, "data"), "the table's data must survive") +} + +// A create that checked the name while it was free must not write over a +// delete's restore: the claim loses instead of layering new metadata over the +// old tags, policy, and maintenance settings. +func TestCreateTableLosesToRestoredMarkers(t *testing.T) { + fs, m := startRenameManager(t) + + nsPath := GetNamespacePath(renameTestBucket, "ns") + fs.Put(nsPath, "newt", nil) + + oldMeta, _ := json.Marshal(tableMetadataInternal{ + Name: "newt", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + }) + restored := map[string][]byte{ + ExtendedKeyMetadata: oldMeta, + ExtendedKeyEntryType: []byte(EntryTypeTable), + ExtendedKeyTags: []byte(`{"old":"tag"}`), + ExtendedKeyMaintenance: []byte(`{"old":"maintenance"}`), + } + fs.BeforeUpdate = func() { + fs.Put(nsPath, "newt", restored) + } + + status, resp := runHandler(t, m, fs.Client, "CreateTable", &CreateTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "newt", + Format: FormatIceberg, + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeTableAlreadyExists, resp.Type) + + got := fs.Get(nsPath, "newt") + require.NotNil(t, got) + assert.Equal(t, restored[ExtendedKeyTags], got.Extended[ExtendedKeyTags], "restored markers must not be overwritten") + assert.Equal(t, restored[ExtendedKeyMetadata], got.Extended[ExtendedKeyMetadata]) +} + // A table whose data was decoupled from its name (created over a leftover from a // rename): catalog marker at ns/newt, data at ns/newt-x, and ns/newt still holds // another table's leftover data. Dropping it must purge only its own data and @@ -168,3 +249,77 @@ func TestDeleteTableRefusesNestedSiblingDataPath(t *testing.T) { assert.NotNil(t, fs.Get(GetNamespacePath(renameTestBucket, "nsa"), "vtab"), "the victim table's catalog entry must survive") } + +// A rename that completes in the token delete's check-to-claim window moves +// the record to another name: the stale delete must lose on the claim instead +// of purging the data the moved record still points at. +func TestDeleteTableTokenDeleteLosesToCompletedRename(t *testing.T) { + fs, m := startRenameManager(t) + + token := "tok-rn" + meta, _ := json.Marshal(tableMetadataInternal{ + Name: "rn", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + VersionToken: token, + MetadataLocation: "s3://" + renameTestBucket + "/ns/old-x/metadata/v1.metadata.json", + }) + fs.Put(GetNamespacePath(renameTestBucket, "ns"), "rn", map[string][]byte{ + ExtendedKeyMetadata: meta, + ExtendedKeyEntryType: []byte(EntryTypeTable), + }) + dataPath := GetTablePath(renameTestBucket, "ns", "old-x") + fs.Put(dataPath, "data", nil) + + fs.BeforeUpdate = func() { + // The rename's source clear has already run by the time the delete's + // claim write lands. + fs.Put(GetNamespacePath(renameTestBucket, "ns"), "rn", nil) + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTable", &DeleteTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "rn", + VersionToken: token, + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeConflict, resp.Type) + + assert.NotNil(t, fs.Get(dataPath, "data"), "the moved table's data must survive the stale delete") +} + +// A delete's claim on the source must turn a rename into a conflict before +// any destination entry exists: the claim is what keeps a losing rename from +// leaving a destination marker pointing at data the delete went on to purge. +func TestRenameTableLosesToDeleteClaim(t *testing.T) { + fs, m := startRenameManager(t) + + nsPath := GetNamespacePath(renameTestBucket, "ns") + fs.BeforeUpdate = func() { + e := fs.Get(nsPath, "t") + claimed := map[string][]byte{} + for k, v := range e.Extended { + claimed[k] = v + } + claimed[ExtendedKeyDeleting] = []byte("delete") + fs.Put(nsPath, "t", claimed) + } + + status, resp := runHandler(t, m, fs.Client, "RenameTable", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeConflict, resp.Type) + + assert.Nil(t, fs.Get(nsPath, "t2"), "a losing rename must not publish the destination") + src := fs.Get(nsPath, "t") + require.NotNil(t, src) + assert.NotNil(t, src.Extended[ExtendedKeyMetadata], "the claimed source record must survive") + assert.NotNil(t, fs.Get(GetTablePath(renameTestBucket, "ns", "t"), "data")) +} diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 5ee6f01391..79c41cf674 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -13,6 +13,7 @@ import ( "strings" "time" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) @@ -36,7 +37,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque return fmt.Errorf("invalid format") } - target, err := h.authorizeCreateTable(w, r, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags) + target, err := h.authorizeCreateTable(w, r, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags, false) if err != nil { return err } @@ -129,6 +130,14 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque return fmt.Errorf("failed to marshal metadata: %w", err) } + var tagsBytes []byte + if len(req.Tags) > 0 { + if tagsBytes, err = json.Marshal(req.Tags); err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal tags") + return fmt.Errorf("failed to marshal tags: %w", err) + } + } + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { // Ensure table directory exists (may already be created by object storage clients) if err := h.ensureDirectory(r.Context(), client, tablePath); err != nil { @@ -141,31 +150,21 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque return err } - // Set metadata as extended attribute - if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes); err != nil { - return err + attrs := map[string][]byte{ + ExtendedKeyMetadata: metadataBytes, + ExtendedKeyEntryType: []byte(EntryTypeTable), } - - // Tag the entry as a table so view listings can exclude it. - if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyEntryType, []byte(EntryTypeTable)); err != nil { - return err + if len(tagsBytes) > 0 { + attrs[ExtendedKeyTags] = tagsBytes } - - // Set tags if provided - if len(req.Tags) > 0 { - tagsBytes, err := json.Marshal(req.Tags) - if err != nil { - return fmt.Errorf("failed to marshal tags: %w", err) - } - if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyTags, tagsBytes); err != nil { - return err - } - } - - return nil + return h.claimCatalogEntry(r.Context(), client, tablePath, attrs) }) if err != nil { + if errors.Is(err, ErrNameTaken) { + h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", tableName)) + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create table") return err } @@ -228,7 +227,7 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req return fmt.Errorf("metadataLocation is required") } - target, err := h.authorizeCreateTable(w, r, filerClient, req.TableBucketARN, req.Namespace, req.Name, nil) + target, err := h.authorizeCreateTable(w, r, filerClient, req.TableBucketARN, req.Namespace, req.Name, nil, false) if err != nil { return err } @@ -282,9 +281,15 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req if err := h.ensureDirectory(r.Context(), client, tablePath); err != nil { return err } - return h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes) + return h.claimCatalogEntry(r.Context(), client, tablePath, map[string][]byte{ + ExtendedKeyMetadata: metadataBytes, + }) }) if err != nil { + if errors.Is(err, ErrNameTaken) { + h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", tableName)) + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to register table") return err } @@ -896,8 +901,16 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque var bucketTags map[string]string var tableTags map[string]string var bucketMetadata tableBucketMetadata + var tableEntry *filer_pb.Entry err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - data, err := h.readTableMetadata(r.Context(), client, tablePath) + // One lookup feeds both the version-token check and the conditional + // delete below: a second read could land on a replacement entry while + // the token was validated against the old one. + tableEntry, err = h.lookupEntry(r.Context(), client, tablePath) + if err != nil { + return err + } + data, err := tableMetadata(tableEntry) if err != nil { return err } @@ -1035,6 +1048,40 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque // lives elsewhere. Purge the data, then clear the catalog marker // without deleting the name path -- it may still hold another // table's data that was left when this name was reused. + if req.VersionToken != "" && tableEntry != nil { + // A token caller owns exactly the record it read. Claim it + // under the captured attributes before any data is removed: + // a rename or delete+recreate that moved the record aborts + // here, while it still points at live data. The claim holds + // the name through the purge, a failed purge releases it so + // the intact record can be retried, and the marker comes off + // under a compare on the claimed record. + snapshot := snapshotExtended(tableEntry.Extended) + if err := h.claimCatalogRecord(r.Context(), client, tablePath, snapshot, "delete"); err != nil { + if errors.Is(err, ErrConcurrentUpdate) { + return ErrVersionTokenMismatch + } + return err + } + claimed := snapshotExtended(tableEntry.Extended) + claimed[ExtendedKeyDeleting] = []byte("delete") + if err := h.deleteDirectory(r.Context(), client, dataPath); err != nil { + if relErr := h.releaseCatalogRecord(r.Context(), client, tablePath, claimed); relErr != nil { + glog.Warningf("delete table %s: release claim: %v", tablePath, relErr) + } + return err + } + err := h.removeExtendedAttributesIf(r.Context(), client, tablePath, claimed, + ExtendedKeyMetadata, ExtendedKeyMetadataVersion, ExtendedKeyPolicy, ExtendedKeyTags, ExtendedKeyEntryType, + ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyDeleting) + if err != nil { + if errors.Is(err, ErrConcurrentUpdate) { + return ErrVersionTokenMismatch + } + return err + } + return nil + } if err := h.deleteDirectory(r.Context(), client, dataPath); err != nil { return err } @@ -1042,11 +1089,27 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque ExtendedKeyMetadata, ExtendedKeyMetadataVersion, ExtendedKeyPolicy, ExtendedKeyTags, ExtendedKeyEntryType, ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus) } - // Colocated table: the name path holds the data. + // Colocated table: the name path holds the data. A token caller's + // delete carries the captured attributes as a filer precondition, so + // compare and removal are one operation under the path lock: an xattr + // write that keeps Mtime, or a delete+recreate in between, refuses it. + if req.VersionToken != "" && tableEntry != nil { + if err := h.deleteDirectoryIfMatches(r.Context(), client, tablePath, tableEntry); err != nil { + if errors.Is(err, ErrConcurrentUpdate) { + return ErrVersionTokenMismatch + } + return err + } + return nil + } return h.deleteDirectory(r.Context(), client, tablePath) }) if err != nil { + if errors.Is(err, ErrVersionTokenMismatch) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "version token mismatch") + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table") return err } @@ -1123,6 +1186,13 @@ func (h *S3TablesHandler) readTableMetadata(ctx context.Context, client filer_pb if err != nil { return nil, err } + return tableMetadata(entry) +} + +// tableMetadata extracts the table-metadata xattr from an entry. A view stored +// at the same name is reported absent, the way GetTable and the renamer's +// entry-type check already treat kind mismatches. +func tableMetadata(entry *filer_pb.Entry) ([]byte, error) { if EntryType(entry.Extended) == EntryTypeView { return nil, filer_pb.ErrNotFound } @@ -1214,10 +1284,6 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ var metadataVersionXattr []byte var maintenanceXattr []byte var maintenanceStatusXattr []byte - // The values the rename copies. The source is only cleared while it still - // holds exactly these, so a write that lands mid-rename is not deleted here - // after having missed the copy to the destination. - var copiedFromSource map[string][]byte var tablePolicy string var bucketPolicy string var bucketTags map[string]string @@ -1255,10 +1321,6 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ return err } srcExtended = srcEntry.Extended - copiedFromSource = make(map[string][]byte, len(renamedTableAttributes)) - for _, key := range renamedTableAttributes { - copiedFromSource[key] = srcEntry.Extended[key] - } // The maintenance configuration and its last-run status belong to the // table, so they move with it; leaving them behind re-enables @@ -1421,52 +1483,69 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ return fmt.Errorf("failed to marshal metadata: %w", err) } - // Write the destination entry before deleting the source so a mid-rename - // failure can never lose the table. + // Claim the source record before publishing the destination: a delete or + // a second rename that wins the claim backs out here — while no + // destination entry exists — instead of leaving one behind pointing at + // data a delete already removed. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - if err := h.createDirectory(r.Context(), client, destPath); err != nil { + if err := h.claimCatalogRecord(r.Context(), client, srcPath, snapshotExtended(srcExtended), "rename"); err != nil { return err } + claimed := snapshotExtended(srcExtended) + claimed[ExtendedKeyDeleting] = []byte("rename") + releaseClaim := func(err error) error { + if relErr := h.releaseCatalogRecord(r.Context(), client, srcPath, claimed); relErr != nil { + glog.Warningf("rename %s: release claim: %v", srcPath, relErr) + } + return err + } + if err := h.createDirectory(r.Context(), client, destPath); err != nil { + return releaseClaim(err) + } if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadata, metadataBytes); err != nil { - return err + return releaseClaim(err) } if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyEntryType, []byte(kind.entryType)); err != nil { - return err + return releaseClaim(err) } if len(metadataVersionXattr) > 0 { if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadataVersion, metadataVersionXattr); err != nil { - return err + return releaseClaim(err) } } if len(tableTags) > 0 { tagsBytes, err := json.Marshal(tableTags) if err != nil { - return fmt.Errorf("failed to marshal tags: %w", err) + return releaseClaim(fmt.Errorf("failed to marshal tags: %w", err)) } if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyTags, tagsBytes); err != nil { - return err + return releaseClaim(err) } } if tablePolicy != "" { if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyPolicy, []byte(tablePolicy)); err != nil { - return err + return releaseClaim(err) } } if len(maintenanceXattr) > 0 { if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMaintenance, maintenanceXattr); err != nil { - return err + return releaseClaim(err) } } if len(maintenanceStatusXattr) > 0 { if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMaintenanceStatus, maintenanceStatusXattr); err != nil { - return err + return releaseClaim(err) } } // Soft-delete the source catalog identity in place: drop its catalog xattrs // so the name stops resolving while the metadata/ and data/ children stay put // (manifests embed absolute paths, so the data must not move). - return h.removeExtendedAttributesIf(r.Context(), client, srcPath, copiedFromSource, - renamedTableAttributes...) + err := h.removeExtendedAttributesIf(r.Context(), client, srcPath, claimed, + append(append([]string{}, renamedTableAttributes...), ExtendedKeyDeleting)...) + if err != nil { + return releaseClaim(err) + } + return nil }) if err != nil { @@ -1750,7 +1829,7 @@ type createTableTarget struct { // Deferred creates (Iceberg stage-create) write into the table bucket before any // table is registered, so they run this same gate first through // Manager.AuthorizeCreateTable. -func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient, tableBucketARN string, namespace []string, name string, requestTags map[string]string) (*createTableTarget, error) { +func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient, tableBucketARN string, namespace []string, name string, requestTags map[string]string, allowMissingNamespace bool) (*createTableTarget, error) { if tableBucketARN == "" { h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "tableBucketARN is required") return nil, fmt.Errorf("tableBucketARN is required") @@ -1781,14 +1860,21 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re // Check if namespace exists namespacePath := GetNamespacePath(bucketName, namespaceName) - namespaceMetadata, err := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName) + nsMetadata, err := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName) + namespaceExists := err == nil if err != nil { - if errors.Is(err, filer_pb.ErrNotFound) { - h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName)) + if errors.Is(err, filer_pb.ErrNotFound) && allowMissingNamespace { + // An implicit namespace can be authorized before it is declared: + // evaluate the grant against the bucket alone. + err = nil } else { - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err)) + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName)) + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err)) + } + return nil, err } - return nil, err } // Authorize table creation using policy framework (namespace + bucket policies) @@ -1811,12 +1897,15 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re } // Fetch namespace policy if it exists - policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy) - if err == nil { - namespacePolicy = string(policyData) - } else if !errors.Is(err, ErrAttributeNotFound) { - return fmt.Errorf("failed to fetch namespace policy: %v", err) + if namespaceExists { + policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy) + if err == nil { + namespacePolicy = string(policyData) + } else if !errors.Is(err, ErrAttributeNotFound) { + return fmt.Errorf("failed to fetch namespace policy: %v", err) + } } + var policyData []byte // Fetch bucket policy if it exists policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy) @@ -1839,6 +1928,11 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re return nil, err } + if !namespaceExists { + // The implicit namespace has no record yet: authorize it as the + // bucket owner's instead of dereferencing the absent metadata. + nsMetadata = &namespaceMetadata{OwnerAccountID: bucketMetadata.OwnerAccountID} + } bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName) identityActions := getIdentityActions(r) policyContext := &PolicyContext{ @@ -1854,7 +1948,7 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re DefaultAllow: h.defaultAllowFor(r), } if !CheckTableOrBucketPermission("CreateTable", accountID, - namespaceMetadata.OwnerAccountID, namespacePolicy, bucketARN, + nsMetadata.OwnerAccountID, namespacePolicy, bucketARN, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, policyContext) { h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table in this namespace") return nil, ErrAccessDenied @@ -1865,6 +1959,6 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re namespaceName: namespaceName, tableName: tableName, bucketFormat: bucketMetadata.Format, - ownerAccountID: namespaceMetadata.OwnerAccountID, + ownerAccountID: nsMetadata.OwnerAccountID, }, nil } diff --git a/weed/s3api/s3tables/handler_view.go b/weed/s3api/s3tables/handler_view.go index 57bad00c5c..f1319c7c07 100644 --- a/weed/s3api/s3tables/handler_view.go +++ b/weed/s3api/s3tables/handler_view.go @@ -151,12 +151,16 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques if err := h.ensureDirectory(r.Context(), client, viewPath); err != nil { return err } - return h.setExtendedAttributes(r.Context(), client, viewPath, map[string][]byte{ + return h.claimCatalogEntry(r.Context(), client, viewPath, map[string][]byte{ ExtendedKeyMetadata: metadataBytes, ExtendedKeyEntryType: []byte(EntryTypeView), }) }) if err != nil { + if errors.Is(err, ErrNameTaken) { + h.writeError(w, http.StatusConflict, ErrCodeViewAlreadyExists, fmt.Sprintf("view %s already exists", viewName)) + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create view") return err } diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index 301be1eb59..da75b5d65f 100644 --- a/weed/s3api/s3tables/manager.go +++ b/weed/s3api/s3tables/manager.go @@ -71,7 +71,7 @@ func (m *Manager) AuthorizeCreateTable(ctx context.Context, filerClient FilerCli return err } recorder := httptest.NewRecorder() - _, authErr := m.handler.authorizeCreateTable(recorder, httpReq, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags) + _, authErr := m.handler.authorizeCreateTable(recorder, httpReq, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags, req.AllowMissingNamespace) if authErr == nil { return nil } diff --git a/weed/s3api/s3tables/s3tablestest/memfiler.go b/weed/s3api/s3tables/s3tablestest/memfiler.go index 5d4540771e..c1047ba9d0 100644 --- a/weed/s3api/s3tables/s3tablestest/memfiler.go +++ b/weed/s3api/s3tables/s3tablestest/memfiler.go @@ -34,6 +34,9 @@ type MemFiler struct { // BeforeUpdate runs once, at the start of the next UpdateEntry, so a test // can land a competing write in a handler's read-to-write window. BeforeUpdate func() + // BeforeDelete runs once before the next DeleteEntry, so a test can land a + // competing write in a handler's check-to-delete window. + BeforeDelete func() // RejectDelete returns the reason DeleteEntry refuses a path, answered the // way the filer answers one: no transport error, the reason in the response. RejectDelete func(dir, name string) string @@ -170,6 +173,10 @@ func (f *MemFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryReque } func (f *MemFiler) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { + if hook := f.BeforeDelete; hook != nil { + f.BeforeDelete = nil + hook() + } if reject := f.RejectDelete; reject != nil { if reason := reject(req.Directory, req.Name); reason != "" { return &filer_pb.DeleteEntryResponse{Error: reason}, nil @@ -177,6 +184,19 @@ func (f *MemFiler) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryReque } f.mu.Lock() defer f.mu.Unlock() + // The real filer evaluates ExpectedExtended under the path lock and refuses + // the delete when the entry no longer matches what the caller captured. + for key, expected := range req.ExpectedExtended { + var actual []byte + if d, ok := f.entries[req.Directory]; ok { + if existing := d[req.Name]; existing != nil { + actual = existing.Extended[key] + } + } + if !bytes.Equal(actual, expected) { + return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key) + } + } if d, ok := f.entries[req.Directory]; ok { delete(d, req.Name) } diff --git a/weed/s3api/s3tables/types.go b/weed/s3api/s3tables/types.go index 748abf62c8..261c724be0 100644 --- a/weed/s3api/s3tables/types.go +++ b/weed/s3api/s3tables/types.go @@ -195,6 +195,10 @@ type CreateTableRequest struct { MetadataVersion int `json:"metadataVersion,omitempty"` MetadataLocation string `json:"metadataLocation,omitempty"` Tags map[string]string `json:"tags,omitempty"` + // AllowMissingNamespace lets an authorization gate evaluate a create whose + // namespace is implicit and not yet declared; the write path still + // validates existence. Never deserialized: internal callers set it. + AllowMissingNamespace bool `json:"-"` } type CreateTableResponse struct { diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 583f51cb90..bc7b08d6bc 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -10,8 +10,12 @@ import ( "strings" "time" + "github.com/google/uuid" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/vector_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/structpb" ) @@ -66,6 +70,9 @@ func arnParts(v *vectorBucketARNs, name, arn string) error { rest := arn[i+len("bucket/"):] parts := strings.Split(rest, "/index/") v.bucket = parts[0] + if v.bucket == "" { + return fmt.Errorf("malformed ARN %q", arn) + } if len(parts) == 2 { v.index = parts[1] } @@ -83,17 +90,16 @@ func resolveIndexARNs(bucketName, bucketArn, indexName, indexArn string) (vector } else if indexArn != "" { return v, fmt.Errorf("indexArn must be an ARN, got %q", indexArn) } - if strings.HasPrefix(bucketArn, "arn:") { + if bucketArn != "" { + if !strings.HasPrefix(bucketArn, "arn:") { + return v, fmt.Errorf("vectorBucketArn must be an ARN, got %q", bucketArn) + } if err := arnParts(&v, "", bucketArn); err != nil { return v, err } } if v.bucket == "" { - if bucketArn != "" { - v.bucket = bucketArn - } else { - v.bucket = bucketName - } + v.bucket = bucketName } if v.index == "" { v.index = indexName @@ -120,46 +126,121 @@ func (s *Server) catalog(ctx context.Context, operation string, req, resp interf return s.tables.Execute(ctx, s.filerClient, operation, req, resp, identityOf(r)) } -// catalogErrCode returns the error code the in-process catalog reports; -// errors from other sources carry no code. +// internalCatalog runs catalog bookkeeping under the trusted identity. It is +// for reads and writes that accompany — never replace — the operation's own +// permission check: bucket format lookups, the implicit index namespace, and +// rollback of a record this request just created. +func (s *Server) internalCatalog(ctx context.Context, operation string, req, resp interface{}) error { + return s.internal.Execute(ctx, s.filerClient, operation, req, resp, "") +} + func catalogErrCode(err error) string { - var e *s3tables.S3TablesError - if errors.As(err, &e) { - return e.Type + var apiErr *s3tables.S3TablesError + if errors.As(err, &apiErr) { + return apiErr.Type } return "" } -func isNotFound(err error) bool { - switch catalogErrCode(err) { - case s3tables.ErrCodeNoSuchBucket, s3tables.ErrCodeNoSuchTable, s3tables.ErrCodeNoSuchNamespace: - return true - } - return err != nil && (strings.Contains(err.Error(), "NoSuchTableBucket") || - strings.Contains(err.Error(), "NoSuchTable") || - strings.Contains(err.Error(), "NotFoundException") || - strings.Contains(err.Error(), "NoSuchBucket")) -} - -func isAlreadyExists(err error) bool { - code := catalogErrCode(err) - return strings.HasSuffix(code, "AlreadyExists") || code == s3tables.ErrCodeConflict || - (err != nil && (strings.Contains(err.Error(), "AlreadyExists") || strings.Contains(err.Error(), "Conflict"))) -} - func catalogErr(w http.ResponseWriter, err error) { - msg := err.Error() - switch { - case isNotFound(err): - writeError(w, http.StatusNotFound, "NotFoundException", msg) - case catalogErrCode(err) == s3tables.ErrCodeAccessDenied || - strings.Contains(msg, "AccessDenied") || strings.Contains(msg, "PermissionDenied"): - writeError(w, http.StatusForbidden, "AccessDeniedException", "access denied") - case isAlreadyExists(err): - writeError(w, http.StatusConflict, "ConflictException", msg) - default: - writeError(w, http.StatusBadRequest, "ValidationException", msg) + var apiErr *s3tables.S3TablesError + if !errors.As(err, &apiErr) { + writeError(w, http.StatusInternalServerError, "InternalFailure", "internal error") + return } + switch { + case apiErr.Type == s3tables.ErrCodeAccessDenied: + writeError(w, http.StatusForbidden, "AccessDeniedException", "access denied") + case strings.HasPrefix(apiErr.Type, "NoSuch"): + writeError(w, http.StatusNotFound, "NotFoundException", apiErr.Message) + case strings.Contains(apiErr.Type, "AlreadyExists") || strings.Contains(apiErr.Type, "NotEmpty") || + apiErr.Type == s3tables.ErrCodeConflict: + writeError(w, http.StatusConflict, "ConflictException", apiErr.Message) + case apiErr.Type == s3tables.ErrCodeInternalError: + writeError(w, http.StatusInternalServerError, "InternalFailure", apiErr.Message) + default: + writeError(w, http.StatusBadRequest, "ValidationException", apiErr.Message) + } +} + +// vectorErr maps a VectorService gRPC status onto the API's error shape. +// Engine messages reach clients only for codes that describe the request +// itself; transport and internal details stay in the server log. +func vectorErr(w http.ResponseWriter, err error) { + msg := status.Convert(err).Message() + switch status.Code(err) { + case codes.NotFound: + writeError(w, http.StatusNotFound, "NotFoundException", msg) + case codes.InvalidArgument, codes.FailedPrecondition, codes.OutOfRange: + writeError(w, http.StatusBadRequest, "ValidationException", msg) + case codes.PermissionDenied, codes.Unauthenticated: + writeError(w, http.StatusForbidden, "AccessDeniedException", "access denied") + case codes.Unavailable: + writeError(w, http.StatusServiceUnavailable, "ServiceUnavailableException", "vector service unavailable") + default: + glog.V(1).Infof("S3 Vectors: engine error: %v", err) + writeError(w, http.StatusInternalServerError, "InternalFailure", "internal error") + } +} + +// lanceBucket reports whether bucket is a table bucket that can hold vector +// indexes: a declared Lance bucket, or one made before formats existed. An +// explicit ICEBERG bucket is not a vector bucket and reads as absent. The +// lookup is internal so the call needs no read grant — the requested +// operation's own permission still gates it. +func (s *Server) lanceBucket(ctx context.Context, bucket string) error { + var resp struct { + Format string `json:"format"` + } + err := s.internalCatalog(ctx, "GetTableBucket", + &s3tables.GetTableBucketRequest{TableBucketARN: s.s3tablesArn(bucket)}, &resp) + if err != nil { + return err + } + if resp.Format != "" && resp.Format != s3tables.FormatLance { + return &s3tables.S3TablesError{ + Type: s3tables.ErrCodeNoSuchBucket, + Message: fmt.Sprintf("vector bucket %s not found", bucket), + } + } + return nil +} + +// ensureIndexNamespace declares the implicit index namespace when absent. The +// calls are internal: a caller's CreateTable grant is what authorizes the +// index, so it must not also need namespace reads or writes. +func (s *Server) ensureIndexNamespace(ctx context.Context, bucket string) error { + var resp map[string]interface{} + err := s.internalCatalog(ctx, "GetNamespace", &s3tables.GetNamespaceRequest{ + TableBucketARN: s.s3tablesArn(bucket), + Namespace: []string{indexNamespace}, + }, &resp) + if err == nil { + return nil + } + if code := catalogErrCode(err); code != s3tables.ErrCodeNoSuchNamespace && code != s3tables.ErrCodeNoSuchBucket { + return err + } + err = s.internalCatalog(ctx, "CreateNamespace", &s3tables.CreateNamespaceRequest{ + TableBucketARN: s.s3tablesArn(bucket), + Namespace: []string{indexNamespace}, + }, &resp) + // A concurrent CreateIndex may have won the create. + if err != nil && catalogErrCode(err) != s3tables.ErrCodeNamespaceAlreadyExists { + return err + } + return nil +} + +// authorizeIndexWrite requires the update grant on the index's catalog table +// — the same check credential vending uses — so a read-only GetTable grant +// never reaches a mutation. +func (s *Server) authorizeIndexWrite(ctx context.Context, r *http.Request, bucket, index string) error { + return s.tables.AuthorizeUpdateTable(ctx, s.filerClient, &s3tables.UpdateTableRequest{ + TableBucketARN: s.s3tablesArn(bucket), + Namespace: []string{indexNamespace}, + Name: index, + }, identityOf(r)) } func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request) { @@ -172,6 +253,13 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request writeError(w, http.StatusBadRequest, "ValidationException", "vectorBucketName is required") return } + // The catalog has no encryption boundary to honor the configuration + // against; accepting it would silently promise semantics we do not keep. + if len(req.EncryptionConfiguration) > 0 { + writeError(w, http.StatusBadRequest, "ValidationException", + "encryptionConfiguration is not supported") + return + } var resp map[string]interface{} err := s.catalog(r.Context(), "CreateTableBucket", &s3tables.CreateTableBucketRequest{Name: req.VectorBucketName, Format: s3tables.FormatLance}, &resp, r) @@ -179,7 +267,7 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request catalogErr(w, err) return } - writeJSON(w, http.StatusOK, &CreateVectorBucketResponse{}) + writeJSON(w, http.StatusOK, &CreateVectorBucketResponse{VectorBucketArn: s.vectorBucketArn(req.VectorBucketName)}) } func (s *Server) handleGetVectorBucket(w http.ResponseWriter, r *http.Request) { @@ -200,6 +288,12 @@ func (s *Server) handleGetVectorBucket(w http.ResponseWriter, r *http.Request) { catalogErr(w, err) return } + // An explicit ICEBERG bucket is not a vector bucket. + if format, _ := resp["format"].(string); format != "" && format != s3tables.FormatLance { + writeError(w, http.StatusNotFound, "NotFoundException", + fmt.Sprintf("vector bucket %s not found", arns.bucket)) + return + } created, _ := time.Parse(time.RFC3339, fmt.Sprint(resp["createdAt"])) writeJSON(w, http.StatusOK, &GetVectorBucketResponse{VectorBucket: &VectorBucket{ VectorBucketArn: s.vectorBucketArn(arns.bucket), @@ -231,10 +325,10 @@ func (s *Server) handleListVectorBuckets(w http.ResponseWriter, r *http.Request) catalogErr(w, err) return } + // The catalog token continues the filtered page: ICEBERG buckets drop out + // of the listing, Lance and format-less legacy buckets stay. out := &ListVectorBucketsResponse{NextToken: resp.ContinuationToken} for _, b := range resp.TableBuckets { - // Buckets pre-dating declared formats accept any format; only a bucket - // that declared another format is excluded here. if b.Format != "" && b.Format != s3tables.FormatLance { continue } @@ -259,15 +353,32 @@ func (s *Server) handleDeleteVectorBucket(w http.ResponseWriter, r *http.Request writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) return } + // The format check needs no read grant — the delete below is the + // authorization boundary — but an ICEBERG bucket is not a vector bucket. + if err := s.lanceBucket(r.Context(), arns.bucket); err != nil { + catalogErr(w, err) + return + } var resp map[string]interface{} - // Indexes live in an implicit namespace; dropping it best-effort lets an - // emptied bucket delete, while surviving indexes keep it non-empty. - _ = s.catalog(r.Context(), "DeleteNamespace", &s3tables.DeleteNamespaceRequest{ - TableBucketARN: s.s3tablesArn(arns.bucket), - Namespace: []string{indexNamespace}, - }, &resp, r) err := s.catalog(r.Context(), "DeleteTableBucket", &s3tables.DeleteTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp, r) + if err != nil && catalogErrCode(err) == s3tables.ErrCodeBucketNotEmpty { + // The implicit index namespace counts as bucket content even when it + // holds no indexes; drop it under the delete's own authorization and + // retry once. A namespace still holding indexes keeps the conflict. + var nsResp map[string]interface{} + nsErr := s.internalCatalog(context.WithoutCancel(r.Context()), "DeleteNamespace", + &s3tables.DeleteNamespaceRequest{ + TableBucketARN: s.s3tablesArn(arns.bucket), + Namespace: []string{indexNamespace}, + }, &nsResp) + if nsErr == nil { + err = s.catalog(r.Context(), "DeleteTableBucket", + &s3tables.DeleteTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp, r) + } else if code := catalogErrCode(nsErr); code != s3tables.ErrCodeNoSuchNamespace { + glog.V(1).Infof("S3 Vectors: drop internal namespace in %s: %v", arns.bucket, nsErr) + } + } if err != nil { catalogErr(w, err) return @@ -315,42 +426,69 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "ValidationException", "dimension must be 1..=4096") return } + if err := s.lanceBucket(r.Context(), arns.bucket); err != nil { + catalogErr(w, err) + return + } + // The create grant is exercised before any write: a denied caller must not + // leave even the implicit namespace behind. The namespace may not exist + // yet, so the check tolerates its absence. + if err := s.tables.AuthorizeCreateTable(r.Context(), s.filerClient, &s3tables.CreateTableRequest{ + TableBucketARN: s.s3tablesArn(arns.bucket), + Namespace: []string{indexNamespace}, + Name: arns.index, + AllowMissingNamespace: true, + }, identityOf(r)); err != nil { + catalogErr(w, err) + return + } + // The namespace is implicit on AWS and internal bookkeeping here. + if err := s.ensureIndexNamespace(r.Context(), arns.bucket); err != nil { + catalogErr(w, err) + return + } + // Only a caller who passed the create check may learn an index exists. + var existing map[string]interface{} + getErr := s.internalCatalog(r.Context(), "GetTable", &s3tables.GetTableRequest{ + TableBucketARN: s.s3tablesArn(arns.bucket), + Namespace: []string{indexNamespace}, + Name: arns.index, + }, &existing) + switch { + case getErr == nil: + writeError(w, http.StatusConflict, "ConflictException", + fmt.Sprintf("index %s already exists", arns.index)) + return + case catalogErrCode(getErr) != s3tables.ErrCodeNoSuchTable: + catalogErr(w, getErr) + return + } metadata := map[string]string{ "dimension": fmt.Sprint(req.Dimension), "distanceMetric": req.DistanceMetric, + // The nonce marks the record this request created, so a rollback after + // an engine failure can never erase a pre-existing index. + "createNonce": uuid.NewString(), } + var nonFilterable []string if req.MetadataConfiguration != nil { - if keys, err := json.Marshal(req.MetadataConfiguration.NonFilterableMetadataKeys); err == nil { + nonFilterable = req.MetadataConfiguration.NonFilterableMetadataKeys + if keys, err := json.Marshal(nonFilterable); err == nil { metadata["nonFilterableMetadataKeys"] = string(keys) } } metadataJSON, _ := json.Marshal(metadata) - var nsResp map[string]interface{} - // The index namespace is implicit on AWS; declare it, tolerating an - // already-existing namespace. - if err := s.catalog(r.Context(), "CreateNamespace", &s3tables.CreateNamespaceRequest{ - TableBucketARN: s.s3tablesArn(arns.bucket), - Namespace: []string{indexNamespace}, - }, &nsResp, r); err != nil && !isAlreadyExists(err) { - catalogErr(w, err) - return - } - var resp map[string]interface{} err = s.catalog(r.Context(), "CreateTable", &s3tables.CreateTableRequest{ TableBucketARN: s.s3tablesArn(arns.bucket), Namespace: []string{indexNamespace}, Name: arns.index, Format: s3tables.FormatLance, Metadata: &s3tables.TableMetadata{FullMetadata: metadataJSON}, - }, &resp, r) + }, nil, r) if err != nil { catalogErr(w, err) return } - var nonFilterable []string - if req.MetadataConfiguration != nil { - nonFilterable = req.MetadataConfiguration.NonFilterableMetadataKeys - } _, err = s.vectors.CreateIndex(r.Context(), &vector_pb.CreateIndexRequest{ Table: tableID(arns.bucket, arns.index), Dimension: uint32(req.Dimension), @@ -359,16 +497,48 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { }) if err != nil { // Roll the index record back so a half-created index does not linger. - var delResp map[string]interface{} - _ = s.catalog(r.Context(), "DeleteTable", &s3tables.DeleteTableRequest{ + // The delete runs past a client disconnect and only removes the record + // this request created: the nonce must match, and the versionToken + // compare-and-swap keeps a racing create from being erased. + ctx := context.WithoutCancel(r.Context()) + var rec map[string]interface{} + if gerr := s.internalCatalog(ctx, "GetTable", &s3tables.GetTableRequest{ TableBucketARN: s.s3tablesArn(arns.bucket), Namespace: []string{indexNamespace}, Name: arns.index, - }, &delResp, r) - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + }, &rec); gerr == nil && recordNonce(rec) == metadata["createNonce"] { + var delResp map[string]interface{} + if derr := s.internalCatalog(ctx, "DeleteTable", &s3tables.DeleteTableRequest{ + TableBucketARN: s.s3tablesArn(arns.bucket), + Namespace: []string{indexNamespace}, + Name: arns.index, + VersionToken: recordVersionToken(rec), + }, &delResp); derr != nil { + glog.Errorf("S3 Vectors: rollback of catalog table %s/%s failed: %v", arns.bucket, arns.index, derr) + } + } else if gerr != nil { + glog.Errorf("S3 Vectors: rollback lookup for %s/%s failed: %v", arns.bucket, arns.index, gerr) + } + vectorErr(w, err) return } - writeJSON(w, http.StatusOK, &CreateIndexResponse{}) + writeJSON(w, http.StatusOK, &CreateIndexResponse{IndexArn: s.indexArn(arns.bucket, arns.index)}) +} + +// recordNonce and recordVersionToken read the catalog record's internal +// bookkeeping for the create-index rollback. +func recordNonce(record map[string]interface{}) string { + if meta, ok := record["metadata"].(map[string]interface{}); ok { + if fm, ok := meta["fullMetadata"].(map[string]interface{}); ok { + return fmt.Sprint(fm["createNonce"]) + } + } + return "" +} + +func recordVersionToken(record map[string]interface{}) string { + token, _ := record["versionToken"].(string) + return token } func indexFromRecord(s *Server, bucket, name string, m map[string]interface{}) *Index { @@ -423,7 +593,7 @@ func (s *Server) handleGetIndex(w http.ResponseWriter, r *http.Request) { catalogErr(w, err) return } - writeJSON(w, http.StatusOK, &GetIndexResponse{Index: indexFromRecord(s, arns.bucket, req.IndexName, resp)}) + writeJSON(w, http.StatusOK, &GetIndexResponse{Index: indexFromRecord(s, arns.bucket, arns.index, resp)}) } func (s *Server) handleListIndexes(w http.ResponseWriter, r *http.Request) { @@ -440,6 +610,7 @@ func (s *Server) handleListIndexes(w http.ResponseWriter, r *http.Request) { var resp struct { TableSummaries []struct { Name string `json:"name"` + Format string `json:"format"` CreatedAt string `json:"createdAt"` } `json:"tables"` ContinuationToken string `json:"continuationToken"` @@ -455,9 +626,21 @@ func (s *Server) handleListIndexes(w http.ResponseWriter, r *http.Request) { catalogErr(w, err) return } + // Summaries carry only what the catalog listing knows; a summary that + // asserted a dimension or metric without a record lookup would lie. out := &ListIndexesResponse{NextToken: resp.ContinuationToken} for _, t := range resp.TableSummaries { - out.Indexes = append(out.Indexes, indexFromRecord(s, arns.bucket, t.Name, nil)) + // A non-Lance table in the index namespace is not an index. + if t.Format != "" && t.Format != s3tables.FormatLance { + continue + } + created, _ := time.Parse(time.RFC3339, t.CreatedAt) + out.Indexes = append(out.Indexes, &IndexSummary{ + IndexName: t.Name, + IndexArn: s.indexArn(arns.bucket, t.Name), + VectorBucketName: arns.bucket, + CreationTime: created, + }) } writeJSON(w, http.StatusOK, out) } @@ -473,9 +656,9 @@ func (s *Server) handleDeleteIndex(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) return } - _, _ = s.vectors.DropIndex(r.Context(), &vector_pb.DropIndexRequest{ - Table: tableID(arns.bucket, arns.index), - }) + // Deleting the catalog record is the delete: the engine derives an index's + // existence from its table, and DropIndex is unimplemented on the shared + // service. var resp map[string]interface{} err = s.catalog(r.Context(), "DeleteTable", &s3tables.DeleteTableRequest{ TableBucketARN: s.s3tablesArn(arns.bucket), @@ -518,12 +701,22 @@ func (s *Server) handlePutVectors(w http.ResponseWriter, r *http.Request) { catalogErr(w, err) return } + // A read grant on the index does not carry a write: mutations require the + // catalog's update grant, the same check credential vending applies. + if err := s.authorizeIndexWrite(r.Context(), r, arns.bucket, arns.index); err != nil { + catalogErr(w, err) + return + } vectors := make([]*vector_pb.Vector, 0, len(req.Vectors)) for _, v := range req.Vectors { if v == nil { writeError(w, http.StatusBadRequest, "ValidationException", "vectors contains a null element") return } + if v.Key == "" { + writeError(w, http.StatusBadRequest, "ValidationException", "vector key must not be empty") + return + } metadata, err := metadataStruct(v.Metadata) if err != nil { writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) @@ -540,7 +733,7 @@ func (s *Server) handlePutVectors(w http.ResponseWriter, r *http.Request) { Vectors: vectors, }) if err != nil { - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + vectorErr(w, err) return } writeJSON(w, http.StatusOK, &PutVectorsResponse{}) @@ -583,7 +776,7 @@ func (s *Server) handleGetVectors(w http.ResponseWriter, r *http.Request) { ReturnMetadata: req.ReturnMetadata, }) if err != nil { - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + vectorErr(w, err) return } writeJSON(w, http.StatusOK, &GetVectorsResponse{ @@ -631,7 +824,7 @@ func (s *Server) handleListVectors(w http.ResponseWriter, r *http.Request) { out, nextToken, err := s.listSegmentVectors(r.Context(), tableID(arns.bucket, arns.index), offset, req.MaxResults, uint64(req.SegmentCount), uint64(req.SegmentIndex), req.ReturnData, req.ReturnMetadata) if err != nil { - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + vectorErr(w, err) return } writeJSON(w, http.StatusOK, &ListVectorsResponse{ @@ -648,7 +841,7 @@ func (s *Server) handleListVectors(w http.ResponseWriter, r *http.Request) { ReturnMetadata: req.ReturnMetadata, }) if err != nil { - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + vectorErr(w, err) return } writeJSON(w, http.StatusOK, &ListVectorsResponse{ @@ -727,12 +920,16 @@ func (s *Server) handleDeleteVectors(w http.ResponseWriter, r *http.Request) { catalogErr(w, err) return } + if err := s.authorizeIndexWrite(r.Context(), r, arns.bucket, arns.index); err != nil { + catalogErr(w, err) + return + } _, err = s.vectors.DeleteVectors(r.Context(), &vector_pb.DeleteVectorsRequest{ Table: tableID(arns.bucket, arns.index), Keys: req.Keys, }) if err != nil { - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + vectorErr(w, err) return } writeJSON(w, http.StatusOK, &DeleteVectorsResponse{}) @@ -787,55 +984,123 @@ func filterExpr(filter map[string]interface{}) (string, error) { return strings.Join(parts, " AND "), nil } +// fieldAccess renders the operand's comparison column over the serialized +// json_extract text. Strings and booleans compare the serialized form +// directly, which is exact-typed and never errors — booleans serialize as +// exactly 'true'/'false', and a cast would also accept the '1'/'0' a stored +// number produces. Numbers go through TRY_CAST, which is NULL on any other +// stored type. The json_get_* extractors are not used — they convert +// numeric-looking strings and booleans, and error on mismatched types. +func fieldAccess(field string, operand interface{}) (string, error) { + path := sqlEscape(existsPath(field)) + switch operand.(type) { + case float64: + return fmt.Sprintf("TRY_CAST(json_extract(`metadata`, '%s') AS DOUBLE)", path), nil + case bool, string: + return fmt.Sprintf("json_extract(`metadata`, '%s')", path), nil + default: + return "", fmt.Errorf("unsupported filter operand %v", operand) + } +} + +// fieldMatch renders "field ". +func fieldMatch(field string, operand interface{}, rhs string) (string, error) { + col, err := fieldAccess(field, operand) + if err != nil { + return "", err + } + return col + " " + rhs, nil +} + +// existsPath renders the field as a JSONPath for the json_exists and +// json_extract parsers: bracket notation keeps dots, spaces, and other +// specials inside the key literal. +func existsPath(field string) string { + field = strings.ReplaceAll(field, `\`, `\\`) + field = strings.ReplaceAll(field, `"`, `\"`) + return `$["` + field + `"]` +} + +func sqlEscape(s string) string { + return strings.ReplaceAll(s, "'", "''") +} + func fieldClause(field string, cond interface{}) (string, error) { - col := fmt.Sprintf("json_get_string(`metadata`, '%s')", strings.ReplaceAll(field, "'", "''")) switch c := cond.(type) { case map[string]interface{}: parts := []string{} for op, operand := range c { switch op { case "$exists": - if operand == true { - parts = append(parts, fmt.Sprintf("json_exists(`metadata`, '%s')", strings.ReplaceAll(field, "'", "''"))) + if b, ok := operand.(bool); ok { + expr := fmt.Sprintf("json_exists(`metadata`, '%s')", sqlEscape(existsPath(field))) + if !b { + expr = "NOT " + expr + } + parts = append(parts, expr) } else { - parts = append(parts, fmt.Sprintf("NOT json_exists(`metadata`, '%s')", strings.ReplaceAll(field, "'", "''"))) + return "", fmt.Errorf("$exists expects a boolean") } case "$eq", "$ne": lit, err := sqlLiteral(operand) if err != nil { return "", err } - sym := "=" - if op == "$ne" { - sym = "<>" + match, err := fieldMatch(field, operand, "= "+lit) + if err != nil { + return "", err } - parts = append(parts, fmt.Sprintf("%s %s %s", col, sym, lit)) + if op == "$ne" { + match = "NOT COALESCE(" + match + ", false)" + } + parts = append(parts, match) case "$in", "$nin": list, ok := operand.([]interface{}) if !ok { return "", fmt.Errorf("%s expects a list", op) } - vals := make([]string, 0, len(list)) + if len(list) == 0 { + if op == "$in" { + parts = append(parts, "false") + } else { + parts = append(parts, "true") + } + continue + } + // Members may mix types; each extractor only sees its own. A + // member whose typed extractor returns NULL must not let NOT + // invert NULL into NULL under $nin, so every group coalesces. + var groups []string for _, item := range list { lit, err := sqlLiteral(item) if err != nil { return "", err } - vals = append(vals, lit) + g, err := fieldMatch(field, item, "in ("+lit+")") + if err != nil { + return "", err + } + groups = append(groups, fmt.Sprintf("COALESCE(%s, false)", g)) } - sym := "IN" + expr := "(" + strings.Join(groups, " OR ") + ")" if op == "$nin" { - sym = "NOT IN" + expr = "NOT " + expr } - parts = append(parts, fmt.Sprintf("%s %s (%s)", col, sym, strings.Join(vals, ", "))) + parts = append(parts, expr) case "$lt", "$lte", "$gt", "$gte": + if _, ok := operand.(float64); !ok { + return "", fmt.Errorf("%s expects a number", op) + } lit, err := sqlLiteral(operand) if err != nil { return "", err } sym := map[string]string{"$lt": "<", "$lte": "<=", "$gt": ">", "$gte": ">="}[op] - numCol := fmt.Sprintf("json_get_float(`metadata`, '%s')", strings.ReplaceAll(field, "'", "''")) - parts = append(parts, fmt.Sprintf("%s %s %s", numCol, sym, lit)) + col, err := fieldAccess(field, operand) + if err != nil { + return "", err + } + parts = append(parts, fmt.Sprintf("%s %s %s", col, sym, lit)) default: return "", fmt.Errorf("unsupported operator %s", op) } @@ -846,19 +1111,50 @@ func fieldClause(field string, cond interface{}) (string, error) { if err != nil { return "", err } - return fmt.Sprintf("%s = %s", col, lit), nil + return fieldMatch(field, cond, "= "+lit) } } +// jsonbString renders s as the JSON text json_extract returns for a stored +// string: quoted, with the same escapes the jsonb serializer emits. +func jsonbString(s string) string { + var b strings.Builder + b.WriteByte('"') + for _, r := range s { + switch r { + case '"': + b.WriteString(`\"`) + case '\\': + b.WriteString(`\\`) + case '\n': + b.WriteString(`\n`) + case '\r': + b.WriteString(`\r`) + case '\t': + b.WriteString(`\t`) + case 0: + b.WriteString(`\0`) + default: + if r < 0x20 || (r >= 0x7f && r <= 0x9f) { + fmt.Fprintf(&b, `\u{%x}`, r) + } else { + b.WriteRune(r) + } + } + } + b.WriteByte('"') + return b.String() +} + func sqlLiteral(v interface{}) (string, error) { switch t := v.(type) { case string: - return "'" + strings.ReplaceAll(t, "'", "''") + "'", nil + return "'" + sqlEscape(jsonbString(t)) + "'", nil case bool: if t { - return "true", nil + return "'true'", nil } - return "false", nil + return "'false'", nil case float64: return fmt.Sprintf("%v", t), nil } @@ -895,8 +1191,8 @@ func (s *Server) handleQueryVectors(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) return } - if req.TopK < 0 { - writeError(w, http.StatusBadRequest, "ValidationException", "topK must be non-negative") + if req.TopK <= 0 { + writeError(w, http.StatusBadRequest, "ValidationException", "topK must be positive") return } resp, err := s.vectors.QueryVectors(r.Context(), &vector_pb.QueryVectorsRequest{ @@ -908,7 +1204,7 @@ func (s *Server) handleQueryVectors(w http.ResponseWriter, r *http.Request) { ReturnDistance: req.ReturnDistance, }) if err != nil { - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + vectorErr(w, err) return } out := &QueryVectorsResponse{DistanceMetric: metric} diff --git a/weed/s3api/s3vectors/handlers_test.go b/weed/s3api/s3vectors/handlers_test.go index 32563acfb7..d20e48a02a 100644 --- a/weed/s3api/s3vectors/handlers_test.go +++ b/weed/s3api/s3vectors/handlers_test.go @@ -155,3 +155,108 @@ func TestListSegmentVectorsPageLimits(t *testing.T) { t.Fatalf("capped page: %d rows token %q", len(out), next) } } + +func TestArnParts(t *testing.T) { + var v vectorBucketARNs + + if err := arnParts(&v, "plain", ""); err != nil || v.bucket != "plain" { + t.Fatalf("plain name: bucket=%q err=%v", v.bucket, err) + } + if err := arnParts(&v, "", "arn:aws:s3vectors:r:a:bucket/mybucket"); err != nil || v.bucket != "mybucket" || v.index != "" { + t.Fatalf("bucket ARN: %+v err=%v", v, err) + } + v = vectorBucketARNs{} + if err := arnParts(&v, "", "arn:aws:s3vectors:r:a:bucket/mybucket/index/myindex"); err != nil || v.bucket != "mybucket" || v.index != "myindex" { + t.Fatalf("index ARN: %+v err=%v", v, err) + } + for _, bad := range []string{ + "arn:aws:s3vectors:r:a:bucket/", + "arn:aws:s3vectors:r:a:bucket//index/i", + "arn:aws:s3vectors:r:a:bucket", + } { + v = vectorBucketARNs{} + if err := arnParts(&v, "", bad); err == nil { + t.Fatalf("ARN %q should be rejected", bad) + } + } +} + +func TestResolveIndexARNs(t *testing.T) { + arns, err := resolveIndexARNs("", "", "", "arn:aws:s3vectors:r:a:bucket/b/index/i") + if err != nil || arns.bucket != "b" || arns.index != "i" { + t.Fatalf("indexArn only: %+v err=%v", arns, err) + } + arns, err = resolveIndexARNs("b", "", "i", "") + if err != nil || arns.bucket != "b" || arns.index != "i" { + t.Fatalf("names only: %+v err=%v", arns, err) + } + if _, err := resolveIndexARNs("", "not-an-arn", "i", ""); err == nil { + t.Fatal("non-ARN vectorBucketArn should be rejected") + } + if _, err := resolveIndexARNs("", "", "", ""); err == nil { + t.Fatal("missing identifiers should be rejected") + } + if _, err := resolveIndexARNs("b", "", "", ""); err == nil { + t.Fatal("missing indexName should be rejected") + } +} + +func TestTranslateFilter(t *testing.T) { + cases := []struct { + name string + filter map[string]interface{} + want string + }{ + {"bare string equality", map[string]interface{}{"k": "v"}, + "json_extract(`metadata`, '$[\"k\"]') = '\"v\"'"}, + {"bare number equality", map[string]interface{}{"k": float64(3)}, + "TRY_CAST(json_extract(`metadata`, '$[\"k\"]') AS DOUBLE) = 3"}, + {"bare bool equality", map[string]interface{}{"k": true}, + "json_extract(`metadata`, '$[\"k\"]') = 'true'"}, + {"ne", map[string]interface{}{"k": map[string]interface{}{"$ne": "v"}}, + "NOT COALESCE(json_extract(`metadata`, '$[\"k\"]') = '\"v\"', false)"}, + {"exists true", map[string]interface{}{"k": map[string]interface{}{"$exists": true}}, + "json_exists(`metadata`, '$[\"k\"]')"}, + {"exists false", map[string]interface{}{"k": map[string]interface{}{"$exists": false}}, + "NOT json_exists(`metadata`, '$[\"k\"]')"}, + {"exists special key", map[string]interface{}{`a.b"c`: map[string]interface{}{"$exists": true}}, + "json_exists(`metadata`, '$[\"a.b\\\"c\"]')"}, + {"range", map[string]interface{}{"k": map[string]interface{}{"$gt": float64(1)}}, + "TRY_CAST(json_extract(`metadata`, '$[\"k\"]') AS DOUBLE) > 1"}, + {"in strings", map[string]interface{}{"k": map[string]interface{}{"$in": []interface{}{"a", "b"}}}, + "(COALESCE(json_extract(`metadata`, '$[\"k\"]') in ('\"a\"'), false) OR COALESCE(json_extract(`metadata`, '$[\"k\"]') in ('\"b\"'), false))"}, + {"nin numbers", map[string]interface{}{"k": map[string]interface{}{"$nin": []interface{}{float64(1)}}}, + "NOT (COALESCE(TRY_CAST(json_extract(`metadata`, '$[\"k\"]') AS DOUBLE) in (1), false))"}, + {"and", map[string]interface{}{"$and": []interface{}{ + map[string]interface{}{"a": "1"}, map[string]interface{}{"b": float64(2)}, + }}, + "(json_extract(`metadata`, '$[\"a\"]') = '\"1\"' AND TRY_CAST(json_extract(`metadata`, '$[\"b\"]') AS DOUBLE) = 2)"}, + {"sql injection", map[string]interface{}{"a'b": "v'1"}, + "json_extract(`metadata`, '$[\"a''b\"]') = '\"v''1\"'"}, + } + for _, tc := range cases { + got, err := translateFilter(tc.filter) + if err != nil { + t.Fatalf("%s: %v", tc.name, err) + } + if got != tc.want { + t.Fatalf("%s:\n got %s\nwant %s", tc.name, got, tc.want) + } + } + + for name, filter := range map[string]map[string]interface{}{ + "unknown operator": {"k": map[string]interface{}{"$bogus": "v"}}, + "exists non-bool": {"k": map[string]interface{}{"$exists": "yes"}}, + "range non-number": {"k": map[string]interface{}{"$gt": "v"}}, + "in non-list": {"k": map[string]interface{}{"$in": "v"}}, + "top-level bad op": {"$bogus": []interface{}{}}, + "nested bad member": {"$and": []interface{}{"x"}}, + } { + if _, err := translateFilter(filter); err == nil { + t.Fatalf("%s should fail", name) + } + } + if got, err := translateFilter(nil); err != nil || got != "" { + t.Fatalf("empty filter: %q %v", got, err) + } +} diff --git a/weed/s3api/s3vectors/server.go b/weed/s3api/s3vectors/server.go index a30f733f07..bcea9907ef 100644 --- a/weed/s3api/s3vectors/server.go +++ b/weed/s3api/s3vectors/server.go @@ -24,10 +24,11 @@ type S3Authenticator interface { DefaultAllow() bool } -// Server implements the AWS S3 Vectors JSON API. +// Server implements the AWS S3 Vectors API. type Server struct { filerClient FilerClient tables *s3tables.Manager + internal *s3tables.Manager vectors vector_pb.VectorServiceClient authenticator S3Authenticator region string @@ -40,9 +41,15 @@ func NewServer(filerClient FilerClient, authenticator S3Authenticator, vectors v if authenticator != nil { manager.SetDefaultAllow(authenticator.DefaultAllow()) } + // Bookkeeping calls that ride along an operation the catalog already + // authorizes — format lookups, the implicit index namespace — must not + // demand read grants the API does not require. + internal := s3tables.NewManager() + internal.SetTrusted(true) return &Server{ filerClient: filerClient, tables: manager, + internal: internal, vectors: vectors, authenticator: authenticator, } diff --git a/weed/s3api/s3vectors/types.go b/weed/s3api/s3vectors/types.go index 78592e28f2..f007262fbf 100644 --- a/weed/s3api/s3vectors/types.go +++ b/weed/s3api/s3vectors/types.go @@ -57,7 +57,9 @@ type CreateVectorBucketRequest struct { MetadataResourceTags map[string]string `json:"-"` } -type CreateVectorBucketResponse struct{} +type CreateVectorBucketResponse struct { + VectorBucketArn string `json:"vectorBucketArn"` +} type GetVectorBucketRequest struct { VectorBucketName string `json:"vectorBucketName"` @@ -107,7 +109,9 @@ type MetadataConfiguration struct { NonFilterableMetadataKeys []string `json:"nonFilterableMetadataKeys"` } -type CreateIndexResponse struct{} +type CreateIndexResponse struct { + IndexArn string `json:"indexArn"` +} type GetIndexRequest struct { IndexArn string `json:"indexArn"` @@ -139,9 +143,18 @@ type ListIndexesRequest struct { VectorBucketArn string `json:"vectorBucketArn"` } +// IndexSummary is the listing shape: the catalog knows the name, ARN, and +// creation time without a per-index metadata read. +type IndexSummary struct { + CreationTime time.Time `json:"creationTime"` + IndexArn string `json:"indexArn"` + IndexName string `json:"indexName"` + VectorBucketName string `json:"vectorBucketName"` +} + type ListIndexesResponse struct { - Indexes []*Index `json:"indexes"` - NextToken string `json:"nextToken,omitempty"` + Indexes []*IndexSummary `json:"indexes"` + NextToken string `json:"nextToken,omitempty"` } type DeleteIndexRequest struct { diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 1c8b8da560..c6295f7810 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -1064,14 +1064,60 @@ func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntr glog.V(4).InfofCtx(ctx, "DeleteEntry %v", req) + fullpath := util.JoinPath(req.Directory, req.Name) + + // A delete carrying extended-attribute preconditions is a read-then-delete + // that the per-path lock below only makes atomic on this filer. Route it + // to the entry's ring owner so one filer's lock arbitrates every writer + // cluster-wide; is_moved bounds this to one hop. + if len(req.ExpectedExtended) > 0 { + if !req.IsMoved { + var ownerResp *filer_pb.DeleteEntryResponse + handled, forwardErr := fs.forwardToWriteOwner(ctx, entryRouteKey(fullpath), func(owner pb.ServerAddress) error { + glog.V(2).InfofCtx(ctx, "DeleteEntry %s: forwarding to owner %s", fullpath, owner) + req.IsMoved = true + return pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + forwarded, e := client.DeleteEntry(ctx, req) + if e != nil { + return e + } + ownerResp = forwarded + return nil + }) + }) + if handled { + if forwardErr != nil { + return &filer_pb.DeleteEntryResponse{}, forwardErr + } + return ownerResp, nil + } + } else { + if err := fs.checkMovedMarker(ctx, req.IsMoved, fs.writeOwner(entryRouteKey(fullpath))); err != nil { + return &filer_pb.DeleteEntryResponse{}, err + } + } + } + // A delete queues the entry's chunks for deletion, so it must not // interleave with a conditional update's check-then-write on the same // path: the update would pass its precondition and then resurrect fids // that are already on the deletion queue. - fullpath := util.JoinPath(req.Directory, req.Name) pathLock := fs.entryLockTable.AcquireLock("DeleteEntry", fullpath, util.ExclusiveLock) defer fs.entryLockTable.ReleaseLock(fullpath, pathLock) + // Extended-attribute preconditions run here, under the lock the delete + // runs under: an attribute write that keeps Mtime can no longer slip + // between the check and the delete. + if len(req.ExpectedExtended) > 0 { + entry, err := fs.filer.FindEntry(ctx, fullpath) + if err != nil { + return &filer_pb.DeleteEntryResponse{}, status.Errorf(codes.FailedPrecondition, "extended attribute precondition %s: %v", fullpath, err) + } + if err := validateUpdateEntryPreconditions(entry, req.ExpectedExtended); err != nil { + return &filer_pb.DeleteEntryResponse{}, err + } + } + if req.KeepRemoteObject { ctx = filer.WithKeepRemoteObject(ctx) }