From 1215ab6d3ec93d0f9d00deb967153dc85fd0a1b0 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 2 Aug 2026 12:24:24 +0800 Subject: [PATCH 1/8] fix: reject ambiguous secondary index names in Go client Validate each secondary index name while applying Put options, before the operation enters the client batch manager. Refs #1265 Signed-off-by: mattisonchao --- oxia/options_base.go | 3 ++- oxia/options_delete.go | 3 ++- oxia/options_list.go | 5 +++-- oxia/options_put.go | 30 +++++++++++++++++++++--------- oxia/options_put_test.go | 32 ++++++++++++++++++++++++++++++++ 5 files changed, 60 insertions(+), 13 deletions(-) create mode 100644 oxia/options_put_test.go diff --git a/oxia/options_base.go b/oxia/options_base.go index c35d77ef7..00e1abbaf 100644 --- a/oxia/options_base.go +++ b/oxia/options_base.go @@ -43,8 +43,9 @@ type partitionKeyOpt struct { partitionKey *string } -func (o *partitionKeyOpt) applyPut(opts *putOptions) { +func (o *partitionKeyOpt) applyPut(opts *putOptions) error { opts.partitionKey = o.partitionKey + return nil } func (o *partitionKeyOpt) applyDelete(opts *deleteOptions) { diff --git a/oxia/options_delete.go b/oxia/options_delete.go index 37e241aab..9846397ee 100644 --- a/oxia/options_delete.go +++ b/oxia/options_delete.go @@ -43,8 +43,9 @@ type expectedVersionId struct { versionId int64 } -func (e *expectedVersionId) applyPut(opts *putOptions) { +func (e *expectedVersionId) applyPut(opts *putOptions) error { opts.expectedVersion = &e.versionId + return nil } func (e *expectedVersionId) applyDelete(opts *deleteOptions) { diff --git a/oxia/options_list.go b/oxia/options_list.go index 0da9695d0..2c3fb0d4d 100644 --- a/oxia/options_list.go +++ b/oxia/options_list.go @@ -52,8 +52,9 @@ func (u *useIndex) applyGet(opts *getOptions) { opts.secondaryIndexName = &u.indexName } -// UseIndex let the users specify a different index to follow for the -// Note: The returned list will contain they primary keys of the records. +// UseIndex selects the secondary index used to interpret keys and key ranges +// in Get, List, and RangeScan operations. Index names must not contain '/'. +// Returned record keys remain the corresponding primary keys. func UseIndex(indexName string) ListOption { return &useIndex{indexName} } diff --git a/oxia/options_put.go b/oxia/options_put.go index bbed45e2e..b982d08a3 100644 --- a/oxia/options_put.go +++ b/oxia/options_put.go @@ -14,7 +14,11 @@ package oxia -import "github.com/pkg/errors" +import ( + "strings" + + "github.com/pkg/errors" +) type putOptions struct { baseOptions @@ -26,13 +30,15 @@ type putOptions struct { // PutOption represents an option for the [SyncClient.Put] operation. type PutOption interface { - applyPut(opts *putOptions) + applyPut(opts *putOptions) error } func newPutOptions(opts []PutOption) (*putOptions, error) { putOpts := &putOptions{} for _, opt := range opts { - opt.applyPut(putOpts) + if err := opt.applyPut(putOpts); err != nil { + return nil, err + } } if len(putOpts.sequenceKeysDeltas) > 0 { @@ -62,8 +68,9 @@ type ephemeral struct{} var ephemeralFlag = &ephemeral{} -func (*ephemeral) applyPut(opts *putOptions) { +func (*ephemeral) applyPut(opts *putOptions) error { opts.ephemeral = true + return nil } // Ephemeral marks the record to be created as an ephemeral record. @@ -82,8 +89,9 @@ type sequenceKeysDeltas struct { sequenceKeysDeltas []uint64 } -func (s *sequenceKeysDeltas) applyPut(opts *putOptions) { +func (s *sequenceKeysDeltas) applyPut(opts *putOptions) error { opts.sequenceKeysDeltas = s.sequenceKeysDeltas + return nil } // SequenceKeysDeltas will request that the final record key to be @@ -102,13 +110,17 @@ type secondaryIdxOption struct { secondaryKey string } -func (s *secondaryIdxOption) applyPut(opts *putOptions) { +func (s *secondaryIdxOption) applyPut(opts *putOptions) error { + if strings.IndexByte(s.indexName, '/') >= 0 { + return errors.Wrapf(ErrInvalidOptions, + "secondary index name %q must not contain '/'", s.indexName) + } opts.secondaryIndexes = append(opts.secondaryIndexes, s) + return nil } -// SecondaryIndex let the users specify additional keys to index the record -// Index names are arbitrary strings and can be used in `List` and -// `RangeScan` requests. +// SecondaryIndex lets users specify additional keys to index the record. +// Index names must not contain '/' and can be used in [UseIndex] options. // Secondary keys are not required to be unique. // Multiple secondary indexes can be passed on the same record, even // reusing multiple times the same indexName. diff --git a/oxia/options_put_test.go b/oxia/options_put_test.go new file mode 100644 index 000000000..8a2d5a9c2 --- /dev/null +++ b/oxia/options_put_test.go @@ -0,0 +1,32 @@ +// Copyright 2023-2026 The Oxia Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package oxia + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSecondaryIndexNameValidation(t *testing.T) { + _, err := newPutOptions([]PutOption{SecondaryIndex("tenant/users", "email")}) + require.Error(t, err) + assert.ErrorIs(t, err, ErrInvalidOptions) + assert.ErrorContains(t, err, "must not contain '/'") + + _, err = newPutOptions([]PutOption{SecondaryIndex("tenant", "users/email")}) + assert.NoError(t, err) +} From cb6b7c9d6c2a7a3ac77771774387ea7f80e9f67e Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 2 Aug 2026 12:59:51 +0800 Subject: [PATCH 2/8] fix: reject invalid secondary index names on servers Signed-off-by: mattisonchao --- common/proto/client.pb.go | 9 +- common/proto/client.proto | 2 + common/proto/replication.pb.go | 16 ++- common/proto/replication.proto | 1 + oxia/options_put_test.go | 7 ++ oxia/proto_utils.go | 2 + oxiad/common/feature/feature.go | 1 + .../shard/shard_controller_election_test.go | 20 ++++ .../controller/shard/shard_controller_test.go | 4 +- oxiad/dataserver/database/db.go | 8 ++ oxiad/dataserver/database/db_test.go | 109 ++++++++++++++++-- 11 files changed, 159 insertions(+), 20 deletions(-) diff --git a/common/proto/client.pb.go b/common/proto/client.pb.go index 83a67453a..03c9f4daa 100644 --- a/common/proto/client.pb.go +++ b/common/proto/client.pb.go @@ -159,6 +159,8 @@ const ( Status_UNEXPECTED_VERSION_ID Status = 2 // The session that the put request referred to is not alive Status_SESSION_DOES_NOT_EXIST Status = 3 + // The request contains an invalid argument + Status_INVALID_ARGUMENT Status = 4 ) // Enum value maps for Status. @@ -168,12 +170,14 @@ var ( 1: "KEY_NOT_FOUND", 2: "UNEXPECTED_VERSION_ID", 3: "SESSION_DOES_NOT_EXIST", + 4: "INVALID_ARGUMENT", } Status_value = map[string]int32{ "OK": 0, "KEY_NOT_FOUND": 1, "UNEXPECTED_VERSION_ID": 2, "SESSION_DOES_NOT_EXIST": 3, + "INVALID_ARGUMENT": 4, } ) @@ -2590,12 +2594,13 @@ const file_client_proto_rawDesc = "" + "\aCEILING\x10\x02\x12\t\n" + "\x05LOWER\x10\x03\x12\n" + "\n" + - "\x06HIGHER\x10\x04*Z\n" + + "\x06HIGHER\x10\x04*p\n" + "\x06Status\x12\x06\n" + "\x02OK\x10\x00\x12\x11\n" + "\rKEY_NOT_FOUND\x10\x01\x12\x19\n" + "\x15UNEXPECTED_VERSION_ID\x10\x02\x12\x1a\n" + - "\x16SESSION_DOES_NOT_EXIST\x10\x03*]\n" + + "\x16SESSION_DOES_NOT_EXIST\x10\x03\x12\x14\n" + + "\x10INVALID_ARGUMENT\x10\x04*]\n" + "\x10NotificationType\x12\x0f\n" + "\vKEY_CREATED\x10\x00\x12\x10\n" + "\fKEY_MODIFIED\x10\x01\x12\x0f\n" + diff --git a/common/proto/client.proto b/common/proto/client.proto index 7cbcd2bfc..0872bc40a 100644 --- a/common/proto/client.proto +++ b/common/proto/client.proto @@ -468,6 +468,8 @@ enum Status { UNEXPECTED_VERSION_ID = 2; // The session that the put request referred to is not alive SESSION_DOES_NOT_EXIST = 3; + // The request contains an invalid argument + INVALID_ARGUMENT = 4; } message CreateSessionRequest { diff --git a/common/proto/replication.pb.go b/common/proto/replication.pb.go index 931befc2c..8bf1635b1 100644 --- a/common/proto/replication.pb.go +++ b/common/proto/replication.pb.go @@ -40,8 +40,9 @@ const ( type Feature int32 const ( - Feature_FEATURE_UNKNOWN Feature = 0 - Feature_FEATURE_DB_CHECKSUM Feature = 1 + Feature_FEATURE_UNKNOWN Feature = 0 + Feature_FEATURE_DB_CHECKSUM Feature = 1 + Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION Feature = 2 ) // Enum value maps for Feature. @@ -49,10 +50,12 @@ var ( Feature_name = map[int32]string{ 0: "FEATURE_UNKNOWN", 1: "FEATURE_DB_CHECKSUM", + 2: "FEATURE_SECONDARY_INDEX_NAME_VALIDATION", } Feature_value = map[string]int32{ - "FEATURE_UNKNOWN": 0, - "FEATURE_DB_CHECKSUM": 1, + "FEATURE_UNKNOWN": 0, + "FEATURE_DB_CHECKSUM": 1, + "FEATURE_SECONDARY_INDEX_NAME_VALIDATION": 2, } ) @@ -1980,10 +1983,11 @@ const file_replication_proto_rawDesc = "" + "ShardStats\x12\"\n" + "\rdb_size_bytes\x18\x01 \x01(\x04R\vdbSizeBytes\x12$\n" + "\x0eread_ops_total\x18\x02 \x01(\x04R\freadOpsTotal\x12&\n" + - "\x0fwrite_ops_total\x18\x03 \x01(\x04R\rwriteOpsTotal*7\n" + + "\x0fwrite_ops_total\x18\x03 \x01(\x04R\rwriteOpsTotal*d\n" + "\aFeature\x12\x13\n" + "\x0fFEATURE_UNKNOWN\x10\x00\x12\x17\n" + - "\x13FEATURE_DB_CHECKSUM\x10\x01*\x8e\x01\n" + + "\x13FEATURE_DB_CHECKSUM\x10\x01\x12+\n" + + "'FEATURE_SECONDARY_INDEX_NAME_VALIDATION\x10\x02*\x8e\x01\n" + "\x0fHandshakeStatus\x12\x1c\n" + "\x18HANDSHAKE_STATUS_UNKNOWN\x10\x00\x12\x1a\n" + "\x16HANDSHAKE_STATUS_BOUND\x10\x01\x12\"\n" + diff --git a/common/proto/replication.proto b/common/proto/replication.proto index 237f16458..8b2a16272 100644 --- a/common/proto/replication.proto +++ b/common/proto/replication.proto @@ -61,6 +61,7 @@ service OxiaLogReplication { enum Feature { FEATURE_UNKNOWN = 0; FEATURE_DB_CHECKSUM = 1; + FEATURE_SECONDARY_INDEX_NAME_VALIDATION = 2; } message GetInfoRequest { diff --git a/oxia/options_put_test.go b/oxia/options_put_test.go index 8a2d5a9c2..db033023e 100644 --- a/oxia/options_put_test.go +++ b/oxia/options_put_test.go @@ -19,6 +19,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/oxia-db/oxia/common/proto" ) func TestSecondaryIndexNameValidation(t *testing.T) { @@ -30,3 +32,8 @@ func TestSecondaryIndexNameValidation(t *testing.T) { _, err = newPutOptions([]PutOption{SecondaryIndex("tenant", "users/email")}) assert.NoError(t, err) } + +func TestInvalidArgumentStatus(t *testing.T) { + result := toPutResult("key", &proto.PutResponse{Status: proto.Status_INVALID_ARGUMENT}) + assert.ErrorIs(t, result.Err, ErrInvalidOptions) +} diff --git a/oxia/proto_utils.go b/oxia/proto_utils.go index 657c7eba3..5348c85bf 100644 --- a/oxia/proto_utils.go +++ b/oxia/proto_utils.go @@ -93,6 +93,8 @@ func toError(status proto.Status) error { return ErrUnexpectedVersionId case proto.Status_KEY_NOT_FOUND: return ErrKeyNotFound + case proto.Status_INVALID_ARGUMENT: + return ErrInvalidOptions default: return ErrUnknownStatus } diff --git a/oxiad/common/feature/feature.go b/oxiad/common/feature/feature.go index 4f5e1fb2d..e54c3eab8 100644 --- a/oxiad/common/feature/feature.go +++ b/oxiad/common/feature/feature.go @@ -24,5 +24,6 @@ import ( func SupportedFeatures() []proto.Feature { return []proto.Feature{ proto.Feature_FEATURE_DB_CHECKSUM, + proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION, } } diff --git a/oxiad/coordinator/runtime/controller/shard/shard_controller_election_test.go b/oxiad/coordinator/runtime/controller/shard/shard_controller_election_test.go index d6ef91b61..7d4b9e357 100644 --- a/oxiad/coordinator/runtime/controller/shard/shard_controller_election_test.go +++ b/oxiad/coordinator/runtime/controller/shard/shard_controller_election_test.go @@ -164,6 +164,26 @@ func TestNegotiate_MixedVersions_RollingUpgrade(t *testing.T) { assert.Contains(t, result, proto.Feature_FEATURE_DB_CHECKSUM, "feature should be enabled after all nodes are upgraded") } +func TestNegotiate_SecondaryIndexValidationRequiresFullUpgrade(t *testing.T) { + currentFeatures := []proto.Feature{ + proto.Feature_FEATURE_DB_CHECKSUM, + proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION, + } + nodeFeatures := map[string][]proto.Feature{ + "new-node-1": currentFeatures, + "new-node-2": currentFeatures, + "old-node": {proto.Feature_FEATURE_DB_CHECKSUM}, + } + + result := negotiate(nodeFeatures, 3) + assert.Contains(t, result, proto.Feature_FEATURE_DB_CHECKSUM) + assert.NotContains(t, result, proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) + + nodeFeatures["old-node"] = currentFeatures + result = negotiate(nodeFeatures, 3) + assert.Contains(t, result, proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) +} + func TestWaitForMajority_Success(t *testing.T) { e := &Election{} server1 := testDataServer("server1") diff --git a/oxiad/coordinator/runtime/controller/shard/shard_controller_test.go b/oxiad/coordinator/runtime/controller/shard/shard_controller_test.go index 516635bc2..c7cff3fb8 100644 --- a/oxiad/coordinator/runtime/controller/shard/shard_controller_test.go +++ b/oxiad/coordinator/runtime/controller/shard/shard_controller_test.go @@ -999,8 +999,8 @@ func TestController_FeatureNegotiation_AllNodesSupport(t *testing.T) { rpc.GetNode(s2).ExpectNewTermRequest(t, shard, 2, true) rpc.GetNode(s3).ExpectNewTermRequest(t, shard, 2, true) - // Verify BecomeLeader includes the DB Checksum feature - rpc.GetNode(s1).ExpectBecomeLeaderRequestWithFeatures(t, shard, 2, 3, []proto.Feature{proto.Feature_FEATURE_DB_CHECKSUM}) + // Verify BecomeLeader includes all features supported by the ensemble. + rpc.GetNode(s1).ExpectBecomeLeaderRequestWithFeatures(t, shard, 2, 3, feature.SupportedFeatures()) assert.Eventually(t, func() bool { return shardStatus(metadata, constant.DefaultNamespace, shard) == proto.ShardStatusSteadyState diff --git a/oxiad/dataserver/database/db.go b/oxiad/dataserver/database/db.go index 6f4196c61..858221825 100644 --- a/oxiad/dataserver/database/db.go +++ b/oxiad/dataserver/database/db.go @@ -723,6 +723,14 @@ func (d *db) ReadTerm() (term int64, options TermOptions, err error) { func (d *db) applyPut(batch kvstore.WriteBatch, baseVersionId *atomic.Int64, notifications *Notifications, putReq *proto.PutRequest, timestamp uint64, updateOperationCallback UpdateOperationCallback, internal bool) (*proto.PutResponse, error) { + if !internal && d.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) { + for _, secondaryIndex := range putReq.SecondaryIndexes { + if strings.IndexByte(secondaryIndex.GetIndexName(), '/') >= 0 { + return &proto.PutResponse{Status: proto.Status_INVALID_ARGUMENT}, nil + } + } + } + var se *proto.StorageEntry var err error var newKey string diff --git a/oxiad/dataserver/database/db_test.go b/oxiad/dataserver/database/db_test.go index a836de1b5..b1b8a801d 100644 --- a/oxiad/dataserver/database/db_test.go +++ b/oxiad/dataserver/database/db_test.go @@ -455,31 +455,120 @@ func TestDB_ReadCommitOffset(t *testing.T) { func TestDB_EnabledFeaturePersistence(t *testing.T) { const commitOffset = int64(7) + for _, enabledFeature := range []proto.Feature{ + proto.Feature_FEATURE_DB_CHECKSUM, + proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION, + } { + t.Run(enabledFeature.String(), func(t *testing.T) { + factory, err := kvstore.NewPebbleKVFactory(kvstore.NewFactoryOptionsForTest(t)) + assert.NoError(t, err) + db, err := NewDB(constant.DefaultNamespace, 1, factory, proto.KeySortingType_NATURAL, 0, time.SystemClock) + assert.NoError(t, err) + + assert.False(t, db.IsFeatureEnabled(enabledFeature)) + + _, err = db.ProcessControlRequest(&proto.ControlRequest{ + Value: &proto.ControlRequest_FeatureEnable{ + FeatureEnable: &proto.FeatureEnableRequest{ + Features: []proto.Feature{enabledFeature}, + }, + }, + }, commitOffset, 0, NoOpCallback) + assert.NoError(t, err) + assert.True(t, db.IsFeatureEnabled(enabledFeature)) + assert.NoError(t, db.Close()) + + db, err = NewDB(constant.DefaultNamespace, 1, factory, proto.KeySortingType_NATURAL, 0, time.SystemClock) + assert.NoError(t, err) + assert.True(t, db.IsFeatureEnabled(enabledFeature)) + + restoredCommitOffset, err := db.ReadCommitOffset() + assert.NoError(t, err) + assert.Equal(t, commitOffset, restoredCommitOffset) + + assert.NoError(t, db.Close()) + assert.NoError(t, factory.Close()) + }) + } +} + +func TestDB_SecondaryIndexNameValidation(t *testing.T) { factory, err := kvstore.NewPebbleKVFactory(kvstore.NewFactoryOptionsForTest(t)) assert.NoError(t, err) db, err := NewDB(constant.DefaultNamespace, 1, factory, proto.KeySortingType_NATURAL, 0, time.SystemClock) assert.NoError(t, err) - assert.False(t, db.IsFeatureEnabled(proto.Feature_FEATURE_DB_CHECKSUM)) + legacyPut := &proto.PutRequest{ + Key: "legacy", + Value: []byte("legacy"), + SecondaryIndexes: []*proto.SecondaryIndex{{ + IndexName: "tenant/users", + SecondaryKey: "email", + }}, + } + response, err := db.ProcessWrite(&proto.WriteRequest{Puts: []*proto.PutRequest{legacyPut}}, 0, 0, NoOpCallback) + assert.NoError(t, err) + assert.Equal(t, proto.Status_OK, response.GetPuts()[0].GetStatus()) _, err = db.ProcessControlRequest(&proto.ControlRequest{ Value: &proto.ControlRequest_FeatureEnable{ FeatureEnable: &proto.FeatureEnableRequest{ - Features: []proto.Feature{proto.Feature_FEATURE_DB_CHECKSUM}, + Features: []proto.Feature{proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION}, }, }, - }, commitOffset, 0, NoOpCallback) + }, 1, 0, NoOpCallback) assert.NoError(t, err) - assert.True(t, db.IsFeatureEnabled(proto.Feature_FEATURE_DB_CHECKSUM)) - assert.NoError(t, db.Close()) - db, err = NewDB(constant.DefaultNamespace, 1, factory, proto.KeySortingType_NATURAL, 0, time.SystemClock) - assert.NoError(t, err) - assert.True(t, db.IsFeatureEnabled(proto.Feature_FEATURE_DB_CHECKSUM)) + callbackPut := &proto.PutRequest{ + Key: FailureCallbackKey, + Value: []byte("invalid"), + SecondaryIndexes: []*proto.SecondaryIndex{{ + IndexName: "tenant/users", + SecondaryKey: "email", + }}, + } + sequencePut := &proto.PutRequest{ + Key: "sequence", + Value: []byte("invalid"), + PartitionKey: pb.String("sequence"), + SequenceKeyDelta: []uint64{1}, + SecondaryIndexes: []*proto.SecondaryIndex{ + {IndexName: "valid", SecondaryKey: "key"}, + {IndexName: "invalid/name", SecondaryKey: "key"}, + }, + } + validPut := &proto.PutRequest{ + Key: "valid", + Value: []byte("valid"), + SecondaryIndexes: []*proto.SecondaryIndex{{ + IndexName: "tenant", + SecondaryKey: "users/email", + }}, + } + response, err = db.ProcessWrite(&proto.WriteRequest{ + Puts: []*proto.PutRequest{callbackPut, sequencePut, validPut}, + }, 2, 0, FailureCallback{}) + assert.NoError(t, err) + assert.Equal(t, proto.Status_INVALID_ARGUMENT, response.GetPuts()[0].GetStatus()) + assert.Equal(t, proto.Status_INVALID_ARGUMENT, response.GetPuts()[1].GetStatus()) + assert.Equal(t, proto.Status_OK, response.GetPuts()[2].GetStatus()) + assert.Equal(t, "sequence", sequencePut.GetKey()) + assert.EqualValues(t, 1, response.GetPuts()[2].GetVersion().GetVersionId()) + + for _, key := range []string{FailureCallbackKey, "sequence"} { + getResponse, getErr := db.Get(&proto.GetRequest{Key: key}) + assert.NoError(t, getErr) + assert.Equal(t, proto.Status_KEY_NOT_FOUND, getResponse.GetStatus()) + } + for _, key := range []string{"legacy", "valid"} { + getResponse, getErr := db.Get(&proto.GetRequest{Key: key}) + assert.NoError(t, getErr) + assert.Equal(t, proto.Status_OK, getResponse.GetStatus()) + } - restoredCommitOffset, err := db.ReadCommitOffset() + commitOffset, err := db.ReadCommitOffset() assert.NoError(t, err) - assert.Equal(t, commitOffset, restoredCommitOffset) + assert.EqualValues(t, 2, commitOffset) assert.NoError(t, db.Close()) assert.NoError(t, factory.Close()) From e3a76c2bcb341c1e5e5cf20b915835fc4e99ecb8 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 2 Aug 2026 13:29:34 +0800 Subject: [PATCH 3/8] refactor: prevalidate puts through update callbacks Signed-off-by: mattisonchao --- .../controller/lead/secondary_indexes.go | 13 ++++++ .../controller/lead/secondary_indexes_test.go | 41 +++++++++++++++++++ .../controller/lead/session_manager.go | 4 ++ oxiad/dataserver/database/db.go | 8 ++-- oxiad/dataserver/database/db_test.go | 10 +++++ oxiad/dataserver/database/noop_callback.go | 4 ++ 6 files changed, 76 insertions(+), 4 deletions(-) diff --git a/oxiad/dataserver/controller/lead/secondary_indexes.go b/oxiad/dataserver/controller/lead/secondary_indexes.go index 52a12f16a..f17871ffc 100644 --- a/oxiad/dataserver/controller/lead/secondary_indexes.go +++ b/oxiad/dataserver/controller/lead/secondary_indexes.go @@ -34,6 +34,10 @@ const secondaryIdxKeyPrefix = constant.InternalKeyPrefix + "idx" type wrapperUpdateCallback struct{} +func (wrapperUpdateCallback) ValidatePut(req *proto.PutRequest) proto.Status { + return secondaryIndexesUpdateCallback.ValidatePut(req) +} + func (wrapperUpdateCallback) OnDeleteWithEntry(batch kvstore.WriteBatch, notifications *database.Notifications, key string, value *proto.StorageEntry) error { // First update the session if err := sessionManagerUpdateOperationCallback.OnDeleteWithEntry(batch, notifications, key, value); err != nil { @@ -81,6 +85,15 @@ type secondaryIndexesUpdateCallbackS struct{} var secondaryIndexesUpdateCallback database.UpdateOperationCallback = &secondaryIndexesUpdateCallbackS{} +func (secondaryIndexesUpdateCallbackS) ValidatePut(request *proto.PutRequest) proto.Status { + for _, secondaryIndex := range request.SecondaryIndexes { + if strings.IndexByte(secondaryIndex.GetIndexName(), '/') >= 0 { + return proto.Status_INVALID_ARGUMENT + } + } + return proto.Status_OK +} + func (secondaryIndexesUpdateCallbackS) OnPut(batch kvstore.WriteBatch, _ *database.Notifications, request *proto.PutRequest, existingEntry *proto.StorageEntry) (proto.Status, error) { if existingEntry != nil { // TODO: We might want to check if there are indexes that did not change diff --git a/oxiad/dataserver/controller/lead/secondary_indexes_test.go b/oxiad/dataserver/controller/lead/secondary_indexes_test.go index d0a98f9c8..31d00aa13 100644 --- a/oxiad/dataserver/controller/lead/secondary_indexes_test.go +++ b/oxiad/dataserver/controller/lead/secondary_indexes_test.go @@ -32,6 +32,47 @@ import ( "github.com/oxia-db/oxia/common/proto" ) +func TestSecondaryIndexNameValidation(t *testing.T) { + tests := []struct { + name string + secondaryIndexes []*proto.SecondaryIndex + expected proto.Status + }{ + {name: "no indexes", expected: proto.Status_OK}, + { + name: "valid index name", + secondaryIndexes: []*proto.SecondaryIndex{{ + IndexName: "tenant", + SecondaryKey: "users/email", + }}, + expected: proto.Status_OK, + }, + { + name: "invalid index name", + secondaryIndexes: []*proto.SecondaryIndex{{ + IndexName: "tenant/users", + SecondaryKey: "email", + }}, + expected: proto.Status_INVALID_ARGUMENT, + }, + { + name: "invalid later index name", + secondaryIndexes: []*proto.SecondaryIndex{ + {IndexName: "tenant", SecondaryKey: "email"}, + {IndexName: "tenant/users", SecondaryKey: "email"}, + }, + expected: proto.Status_INVALID_ARGUMENT, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + request := &proto.PutRequest{SecondaryIndexes: tt.secondaryIndexes} + assert.Equal(t, tt.expected, WrapperUpdateOperationCallback.ValidatePut(request)) + }) + } +} + func TestSecondaryIndices_List(t *testing.T) { var shard int64 = 1 diff --git a/oxiad/dataserver/controller/lead/session_manager.go b/oxiad/dataserver/controller/lead/session_manager.go index e8163ec5b..a6a382f49 100644 --- a/oxiad/dataserver/controller/lead/session_manager.go +++ b/oxiad/dataserver/controller/lead/session_manager.go @@ -388,6 +388,10 @@ type sessionManagerUpdateOperationCallbackS struct{} var sessionManagerUpdateOperationCallback database.UpdateOperationCallback = &sessionManagerUpdateOperationCallbackS{} +func (*sessionManagerUpdateOperationCallbackS) ValidatePut(*proto.PutRequest) proto.Status { + return proto.Status_OK +} + func (*sessionManagerUpdateOperationCallbackS) OnPutWithinSession(batch kvstore.WriteBatch, notification *database.Notifications, request *proto.PutRequest, existingEntry *proto.StorageEntry) (proto.Status, error) { var _, closer, err = batch.Get(SessionKey(SessionId(*request.SessionId))) if err != nil { diff --git a/oxiad/dataserver/database/db.go b/oxiad/dataserver/database/db.go index 858221825..755f27e1d 100644 --- a/oxiad/dataserver/database/db.go +++ b/oxiad/dataserver/database/db.go @@ -61,6 +61,8 @@ const ( ) type UpdateOperationCallback interface { + // ValidatePut must not mutate the request or database state. + ValidatePut(req *proto.PutRequest) proto.Status OnPut(batch kvstore.WriteBatch, notifications *Notifications, req *proto.PutRequest, se *proto.StorageEntry) (proto.Status, error) OnDelete(batch kvstore.WriteBatch, notifications *Notifications, key string) error OnDeleteWithEntry(batch kvstore.WriteBatch, notifications *Notifications, key string, value *proto.StorageEntry) error @@ -724,10 +726,8 @@ func (d *db) applyPut(batch kvstore.WriteBatch, baseVersionId *atomic.Int64, not putReq *proto.PutRequest, timestamp uint64, updateOperationCallback UpdateOperationCallback, internal bool) (*proto.PutResponse, error) { if !internal && d.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) { - for _, secondaryIndex := range putReq.SecondaryIndexes { - if strings.IndexByte(secondaryIndex.GetIndexName(), '/') >= 0 { - return &proto.PutResponse{Status: proto.Status_INVALID_ARGUMENT}, nil - } + if status := updateOperationCallback.ValidatePut(putReq); status != proto.Status_OK { + return &proto.PutResponse{Status: status}, nil } } diff --git a/oxiad/dataserver/database/db_test.go b/oxiad/dataserver/database/db_test.go index b1b8a801d..9c1638481 100644 --- a/oxiad/dataserver/database/db_test.go +++ b/oxiad/dataserver/database/db_test.go @@ -16,6 +16,7 @@ package database import ( "fmt" + "strings" "testing" "github.com/pkg/errors" @@ -1070,6 +1071,15 @@ type FailureCallback struct{} const FailureCallbackKey = "failure" +func (FailureCallback) ValidatePut(req *proto.PutRequest) proto.Status { + for _, secondaryIndex := range req.SecondaryIndexes { + if strings.IndexByte(secondaryIndex.GetIndexName(), '/') >= 0 { + return proto.Status_INVALID_ARGUMENT + } + } + return proto.Status_OK +} + func (f FailureCallback) OnPut(_ kvstore.WriteBatch, _ *Notifications, req *proto.PutRequest, _ *proto.StorageEntry) (proto.Status, error) { if req.Key == FailureCallbackKey { return proto.Status_SESSION_DOES_NOT_EXIST, errors.New("failure injection") diff --git a/oxiad/dataserver/database/noop_callback.go b/oxiad/dataserver/database/noop_callback.go index 18980df16..bff9b260e 100644 --- a/oxiad/dataserver/database/noop_callback.go +++ b/oxiad/dataserver/database/noop_callback.go @@ -21,6 +21,10 @@ import ( type noopCallback struct{} +func (*noopCallback) ValidatePut(*proto.PutRequest) proto.Status { + return proto.Status_OK +} + func (*noopCallback) OnDeleteWithEntry(kvstore.WriteBatch, *Notifications, string, *proto.StorageEntry) error { return nil } From 0d91785347850223853d1ac53b3a61584d3ed55c Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 2 Aug 2026 13:47:56 +0800 Subject: [PATCH 4/8] refactor: move put feature gating into callbacks Signed-off-by: mattisonchao --- .../controller/lead/secondary_indexes.go | 10 +++-- .../controller/lead/secondary_indexes_test.go | 39 +++++++++++++++---- .../controller/lead/session_manager.go | 2 +- oxiad/dataserver/database/db.go | 10 +++-- oxiad/dataserver/database/db_test.go | 8 +++- oxiad/dataserver/database/noop_callback.go | 2 +- 6 files changed, 53 insertions(+), 18 deletions(-) diff --git a/oxiad/dataserver/controller/lead/secondary_indexes.go b/oxiad/dataserver/controller/lead/secondary_indexes.go index f17871ffc..282269a0e 100644 --- a/oxiad/dataserver/controller/lead/secondary_indexes.go +++ b/oxiad/dataserver/controller/lead/secondary_indexes.go @@ -34,8 +34,8 @@ const secondaryIdxKeyPrefix = constant.InternalKeyPrefix + "idx" type wrapperUpdateCallback struct{} -func (wrapperUpdateCallback) ValidatePut(req *proto.PutRequest) proto.Status { - return secondaryIndexesUpdateCallback.ValidatePut(req) +func (wrapperUpdateCallback) ValidatePut(req *proto.PutRequest, features database.FeatureChecker) proto.Status { + return secondaryIndexesUpdateCallback.ValidatePut(req, features) } func (wrapperUpdateCallback) OnDeleteWithEntry(batch kvstore.WriteBatch, notifications *database.Notifications, key string, value *proto.StorageEntry) error { @@ -85,7 +85,11 @@ type secondaryIndexesUpdateCallbackS struct{} var secondaryIndexesUpdateCallback database.UpdateOperationCallback = &secondaryIndexesUpdateCallbackS{} -func (secondaryIndexesUpdateCallbackS) ValidatePut(request *proto.PutRequest) proto.Status { +func (secondaryIndexesUpdateCallbackS) ValidatePut(request *proto.PutRequest, features database.FeatureChecker) proto.Status { + if !features.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) { + return proto.Status_OK + } + for _, secondaryIndex := range request.SecondaryIndexes { if strings.IndexByte(secondaryIndex.GetIndexName(), '/') >= 0 { return proto.Status_INVALID_ARGUMENT diff --git a/oxiad/dataserver/controller/lead/secondary_indexes_test.go b/oxiad/dataserver/controller/lead/secondary_indexes_test.go index 31d00aa13..528dc6332 100644 --- a/oxiad/dataserver/controller/lead/secondary_indexes_test.go +++ b/oxiad/dataserver/controller/lead/secondary_indexes_test.go @@ -25,6 +25,7 @@ import ( "github.com/oxia-db/oxia/oxiad/dataserver/option" "github.com/oxia-db/oxia/common/rpc" + "github.com/oxia-db/oxia/oxiad/dataserver/database" "github.com/oxia-db/oxia/oxiad/dataserver/database/kvstore" "github.com/oxia-db/oxia/common/constant" @@ -34,13 +35,22 @@ import ( func TestSecondaryIndexNameValidation(t *testing.T) { tests := []struct { - name string - secondaryIndexes []*proto.SecondaryIndex - expected proto.Status + name string + validationEnabled bool + secondaryIndexes []*proto.SecondaryIndex + expected proto.Status }{ - {name: "no indexes", expected: proto.Status_OK}, { - name: "valid index name", + name: "validation disabled", + secondaryIndexes: []*proto.SecondaryIndex{{ + IndexName: "tenant/users", + }}, + expected: proto.Status_OK, + }, + {name: "no indexes", validationEnabled: true, expected: proto.Status_OK}, + { + name: "valid index name", + validationEnabled: true, secondaryIndexes: []*proto.SecondaryIndex{{ IndexName: "tenant", SecondaryKey: "users/email", @@ -48,7 +58,8 @@ func TestSecondaryIndexNameValidation(t *testing.T) { expected: proto.Status_OK, }, { - name: "invalid index name", + name: "invalid index name", + validationEnabled: true, secondaryIndexes: []*proto.SecondaryIndex{{ IndexName: "tenant/users", SecondaryKey: "email", @@ -56,7 +67,8 @@ func TestSecondaryIndexNameValidation(t *testing.T) { expected: proto.Status_INVALID_ARGUMENT, }, { - name: "invalid later index name", + name: "invalid later index name", + validationEnabled: true, secondaryIndexes: []*proto.SecondaryIndex{ {IndexName: "tenant", SecondaryKey: "email"}, {IndexName: "tenant/users", SecondaryKey: "email"}, @@ -68,11 +80,22 @@ func TestSecondaryIndexNameValidation(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { request := &proto.PutRequest{SecondaryIndexes: tt.secondaryIndexes} - assert.Equal(t, tt.expected, WrapperUpdateOperationCallback.ValidatePut(request)) + features := testFeatureChecker{secondaryIndexNameValidation: tt.validationEnabled} + assert.Equal(t, tt.expected, WrapperUpdateOperationCallback.ValidatePut(request, features)) }) } } +type testFeatureChecker struct { + secondaryIndexNameValidation bool +} + +func (f testFeatureChecker) IsFeatureEnabled(feature proto.Feature) bool { + return f.secondaryIndexNameValidation && feature == proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION +} + +var _ database.FeatureChecker = testFeatureChecker{} + func TestSecondaryIndices_List(t *testing.T) { var shard int64 = 1 diff --git a/oxiad/dataserver/controller/lead/session_manager.go b/oxiad/dataserver/controller/lead/session_manager.go index a6a382f49..408a007b9 100644 --- a/oxiad/dataserver/controller/lead/session_manager.go +++ b/oxiad/dataserver/controller/lead/session_manager.go @@ -388,7 +388,7 @@ type sessionManagerUpdateOperationCallbackS struct{} var sessionManagerUpdateOperationCallback database.UpdateOperationCallback = &sessionManagerUpdateOperationCallbackS{} -func (*sessionManagerUpdateOperationCallbackS) ValidatePut(*proto.PutRequest) proto.Status { +func (*sessionManagerUpdateOperationCallbackS) ValidatePut(*proto.PutRequest, database.FeatureChecker) proto.Status { return proto.Status_OK } diff --git a/oxiad/dataserver/database/db.go b/oxiad/dataserver/database/db.go index 755f27e1d..8eb6280c7 100644 --- a/oxiad/dataserver/database/db.go +++ b/oxiad/dataserver/database/db.go @@ -60,9 +60,13 @@ const ( termOptionsKey = termKey + "-options" ) +type FeatureChecker interface { + IsFeatureEnabled(feature proto.Feature) bool +} + type UpdateOperationCallback interface { // ValidatePut must not mutate the request or database state. - ValidatePut(req *proto.PutRequest) proto.Status + ValidatePut(req *proto.PutRequest, features FeatureChecker) proto.Status OnPut(batch kvstore.WriteBatch, notifications *Notifications, req *proto.PutRequest, se *proto.StorageEntry) (proto.Status, error) OnDelete(batch kvstore.WriteBatch, notifications *Notifications, key string) error OnDeleteWithEntry(batch kvstore.WriteBatch, notifications *Notifications, key string, value *proto.StorageEntry) error @@ -725,8 +729,8 @@ func (d *db) ReadTerm() (term int64, options TermOptions, err error) { func (d *db) applyPut(batch kvstore.WriteBatch, baseVersionId *atomic.Int64, notifications *Notifications, putReq *proto.PutRequest, timestamp uint64, updateOperationCallback UpdateOperationCallback, internal bool) (*proto.PutResponse, error) { - if !internal && d.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) { - if status := updateOperationCallback.ValidatePut(putReq); status != proto.Status_OK { + if !internal { + if status := updateOperationCallback.ValidatePut(putReq, d); status != proto.Status_OK { return &proto.PutResponse{Status: status}, nil } } diff --git a/oxiad/dataserver/database/db_test.go b/oxiad/dataserver/database/db_test.go index 9c1638481..baa244648 100644 --- a/oxiad/dataserver/database/db_test.go +++ b/oxiad/dataserver/database/db_test.go @@ -507,7 +507,7 @@ func TestDB_SecondaryIndexNameValidation(t *testing.T) { SecondaryKey: "email", }}, } - response, err := db.ProcessWrite(&proto.WriteRequest{Puts: []*proto.PutRequest{legacyPut}}, 0, 0, NoOpCallback) + response, err := db.ProcessWrite(&proto.WriteRequest{Puts: []*proto.PutRequest{legacyPut}}, 0, 0, FailureCallback{}) assert.NoError(t, err) assert.Equal(t, proto.Status_OK, response.GetPuts()[0].GetStatus()) @@ -1071,7 +1071,11 @@ type FailureCallback struct{} const FailureCallbackKey = "failure" -func (FailureCallback) ValidatePut(req *proto.PutRequest) proto.Status { +func (FailureCallback) ValidatePut(req *proto.PutRequest, features FeatureChecker) proto.Status { + if !features.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) { + return proto.Status_OK + } + for _, secondaryIndex := range req.SecondaryIndexes { if strings.IndexByte(secondaryIndex.GetIndexName(), '/') >= 0 { return proto.Status_INVALID_ARGUMENT diff --git a/oxiad/dataserver/database/noop_callback.go b/oxiad/dataserver/database/noop_callback.go index bff9b260e..dcea006e7 100644 --- a/oxiad/dataserver/database/noop_callback.go +++ b/oxiad/dataserver/database/noop_callback.go @@ -21,7 +21,7 @@ import ( type noopCallback struct{} -func (*noopCallback) ValidatePut(*proto.PutRequest) proto.Status { +func (*noopCallback) ValidatePut(*proto.PutRequest, FeatureChecker) proto.Status { return proto.Status_OK } From 41bab962c596177d445035b079c7b8da19f87bdb Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 2 Aug 2026 13:51:15 +0800 Subject: [PATCH 5/8] refactor: always run put prevalidation Signed-off-by: mattisonchao --- oxiad/dataserver/database/db.go | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/oxiad/dataserver/database/db.go b/oxiad/dataserver/database/db.go index 8eb6280c7..c454bcc77 100644 --- a/oxiad/dataserver/database/db.go +++ b/oxiad/dataserver/database/db.go @@ -729,10 +729,8 @@ func (d *db) ReadTerm() (term int64, options TermOptions, err error) { func (d *db) applyPut(batch kvstore.WriteBatch, baseVersionId *atomic.Int64, notifications *Notifications, putReq *proto.PutRequest, timestamp uint64, updateOperationCallback UpdateOperationCallback, internal bool) (*proto.PutResponse, error) { - if !internal { - if status := updateOperationCallback.ValidatePut(putReq, d); status != proto.Status_OK { - return &proto.PutResponse{Status: status}, nil - } + if status := updateOperationCallback.ValidatePut(putReq, d); status != proto.Status_OK { + return &proto.PutResponse{Status: status}, nil } var se *proto.StorageEntry From c0dd124de9b3c1ad25de6980d6809daf67ef3525 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 2 Aug 2026 13:59:31 +0800 Subject: [PATCH 6/8] refactor: move feature checker to feature package Signed-off-by: mattisonchao --- oxiad/common/feature/feature.go | 4 ++++ oxiad/dataserver/controller/lead/secondary_indexes.go | 5 +++-- .../dataserver/controller/lead/secondary_indexes_test.go | 8 ++++---- oxiad/dataserver/controller/lead/session_manager.go | 3 ++- oxiad/dataserver/database/db.go | 7 ++----- oxiad/dataserver/database/db_test.go | 3 ++- oxiad/dataserver/database/noop_callback.go | 3 ++- 7 files changed, 19 insertions(+), 14 deletions(-) diff --git a/oxiad/common/feature/feature.go b/oxiad/common/feature/feature.go index e54c3eab8..5ffc50358 100644 --- a/oxiad/common/feature/feature.go +++ b/oxiad/common/feature/feature.go @@ -21,6 +21,10 @@ import ( "github.com/oxia-db/oxia/common/proto" ) +type Checker interface { + IsFeatureEnabled(feature proto.Feature) bool +} + func SupportedFeatures() []proto.Feature { return []proto.Feature{ proto.Feature_FEATURE_DB_CHECKSUM, diff --git a/oxiad/dataserver/controller/lead/secondary_indexes.go b/oxiad/dataserver/controller/lead/secondary_indexes.go index 282269a0e..d15cb5ee2 100644 --- a/oxiad/dataserver/controller/lead/secondary_indexes.go +++ b/oxiad/dataserver/controller/lead/secondary_indexes.go @@ -24,6 +24,7 @@ import ( "github.com/oxia-db/oxia/common/compare" "github.com/oxia-db/oxia/common/constant" + "github.com/oxia-db/oxia/oxiad/common/feature" "github.com/oxia-db/oxia/oxiad/dataserver/database" "github.com/oxia-db/oxia/oxiad/dataserver/database/kvstore" @@ -34,7 +35,7 @@ const secondaryIdxKeyPrefix = constant.InternalKeyPrefix + "idx" type wrapperUpdateCallback struct{} -func (wrapperUpdateCallback) ValidatePut(req *proto.PutRequest, features database.FeatureChecker) proto.Status { +func (wrapperUpdateCallback) ValidatePut(req *proto.PutRequest, features feature.Checker) proto.Status { return secondaryIndexesUpdateCallback.ValidatePut(req, features) } @@ -85,7 +86,7 @@ type secondaryIndexesUpdateCallbackS struct{} var secondaryIndexesUpdateCallback database.UpdateOperationCallback = &secondaryIndexesUpdateCallbackS{} -func (secondaryIndexesUpdateCallbackS) ValidatePut(request *proto.PutRequest, features database.FeatureChecker) proto.Status { +func (secondaryIndexesUpdateCallbackS) ValidatePut(request *proto.PutRequest, features feature.Checker) proto.Status { if !features.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) { return proto.Status_OK } diff --git a/oxiad/dataserver/controller/lead/secondary_indexes_test.go b/oxiad/dataserver/controller/lead/secondary_indexes_test.go index 528dc6332..1081df822 100644 --- a/oxiad/dataserver/controller/lead/secondary_indexes_test.go +++ b/oxiad/dataserver/controller/lead/secondary_indexes_test.go @@ -25,7 +25,7 @@ import ( "github.com/oxia-db/oxia/oxiad/dataserver/option" "github.com/oxia-db/oxia/common/rpc" - "github.com/oxia-db/oxia/oxiad/dataserver/database" + "github.com/oxia-db/oxia/oxiad/common/feature" "github.com/oxia-db/oxia/oxiad/dataserver/database/kvstore" "github.com/oxia-db/oxia/common/constant" @@ -90,11 +90,11 @@ type testFeatureChecker struct { secondaryIndexNameValidation bool } -func (f testFeatureChecker) IsFeatureEnabled(feature proto.Feature) bool { - return f.secondaryIndexNameValidation && feature == proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION +func (f testFeatureChecker) IsFeatureEnabled(candidate proto.Feature) bool { + return f.secondaryIndexNameValidation && candidate == proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION } -var _ database.FeatureChecker = testFeatureChecker{} +var _ feature.Checker = testFeatureChecker{} func TestSecondaryIndices_List(t *testing.T) { var shard int64 = 1 diff --git a/oxiad/dataserver/controller/lead/session_manager.go b/oxiad/dataserver/controller/lead/session_manager.go index 408a007b9..ec90407d7 100644 --- a/oxiad/dataserver/controller/lead/session_manager.go +++ b/oxiad/dataserver/controller/lead/session_manager.go @@ -27,6 +27,7 @@ import ( "github.com/pkg/errors" + "github.com/oxia-db/oxia/oxiad/common/feature" "github.com/oxia-db/oxia/oxiad/dataserver/database" "github.com/oxia-db/oxia/oxiad/dataserver/database/kvstore" @@ -388,7 +389,7 @@ type sessionManagerUpdateOperationCallbackS struct{} var sessionManagerUpdateOperationCallback database.UpdateOperationCallback = &sessionManagerUpdateOperationCallbackS{} -func (*sessionManagerUpdateOperationCallbackS) ValidatePut(*proto.PutRequest, database.FeatureChecker) proto.Status { +func (*sessionManagerUpdateOperationCallbackS) ValidatePut(*proto.PutRequest, feature.Checker) proto.Status { return proto.Status_OK } diff --git a/oxiad/dataserver/database/db.go b/oxiad/dataserver/database/db.go index c454bcc77..b52a1d8ac 100644 --- a/oxiad/dataserver/database/db.go +++ b/oxiad/dataserver/database/db.go @@ -31,6 +31,7 @@ import ( "go.uber.org/multierr" "github.com/oxia-db/oxia/oxiad/common/crc" + featurepkg "github.com/oxia-db/oxia/oxiad/common/feature" "github.com/oxia-db/oxia/oxiad/dataserver/database/kvstore" @@ -60,13 +61,9 @@ const ( termOptionsKey = termKey + "-options" ) -type FeatureChecker interface { - IsFeatureEnabled(feature proto.Feature) bool -} - type UpdateOperationCallback interface { // ValidatePut must not mutate the request or database state. - ValidatePut(req *proto.PutRequest, features FeatureChecker) proto.Status + ValidatePut(req *proto.PutRequest, features featurepkg.Checker) proto.Status OnPut(batch kvstore.WriteBatch, notifications *Notifications, req *proto.PutRequest, se *proto.StorageEntry) (proto.Status, error) OnDelete(batch kvstore.WriteBatch, notifications *Notifications, key string) error OnDeleteWithEntry(batch kvstore.WriteBatch, notifications *Notifications, key string, value *proto.StorageEntry) error diff --git a/oxiad/dataserver/database/db_test.go b/oxiad/dataserver/database/db_test.go index baa244648..a95b5515d 100644 --- a/oxiad/dataserver/database/db_test.go +++ b/oxiad/dataserver/database/db_test.go @@ -24,6 +24,7 @@ import ( pb "google.golang.org/protobuf/proto" "github.com/oxia-db/oxia/oxiad/common/crc" + "github.com/oxia-db/oxia/oxiad/common/feature" "github.com/oxia-db/oxia/oxiad/dataserver/database/kvstore" @@ -1071,7 +1072,7 @@ type FailureCallback struct{} const FailureCallbackKey = "failure" -func (FailureCallback) ValidatePut(req *proto.PutRequest, features FeatureChecker) proto.Status { +func (FailureCallback) ValidatePut(req *proto.PutRequest, features feature.Checker) proto.Status { if !features.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) { return proto.Status_OK } diff --git a/oxiad/dataserver/database/noop_callback.go b/oxiad/dataserver/database/noop_callback.go index dcea006e7..e84ca2c82 100644 --- a/oxiad/dataserver/database/noop_callback.go +++ b/oxiad/dataserver/database/noop_callback.go @@ -16,12 +16,13 @@ package database import ( "github.com/oxia-db/oxia/common/proto" + "github.com/oxia-db/oxia/oxiad/common/feature" "github.com/oxia-db/oxia/oxiad/dataserver/database/kvstore" ) type noopCallback struct{} -func (*noopCallback) ValidatePut(*proto.PutRequest, FeatureChecker) proto.Status { +func (*noopCallback) ValidatePut(*proto.PutRequest, feature.Checker) proto.Status { return proto.Status_OK } From f64f3d134929a68594cfee69e2eb2b807af3caf6 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 2 Aug 2026 14:06:25 +0800 Subject: [PATCH 7/8] fix: preserve put option compatibility Signed-off-by: mattisonchao --- oxia/options_base.go | 3 +-- oxia/options_delete.go | 3 +-- oxia/options_put.go | 24 +++++++++---------- .../controller/lead/secondary_indexes.go | 4 ++++ 4 files changed, 17 insertions(+), 17 deletions(-) diff --git a/oxia/options_base.go b/oxia/options_base.go index 00e1abbaf..c35d77ef7 100644 --- a/oxia/options_base.go +++ b/oxia/options_base.go @@ -43,9 +43,8 @@ type partitionKeyOpt struct { partitionKey *string } -func (o *partitionKeyOpt) applyPut(opts *putOptions) error { +func (o *partitionKeyOpt) applyPut(opts *putOptions) { opts.partitionKey = o.partitionKey - return nil } func (o *partitionKeyOpt) applyDelete(opts *deleteOptions) { diff --git a/oxia/options_delete.go b/oxia/options_delete.go index 9846397ee..37e241aab 100644 --- a/oxia/options_delete.go +++ b/oxia/options_delete.go @@ -43,9 +43,8 @@ type expectedVersionId struct { versionId int64 } -func (e *expectedVersionId) applyPut(opts *putOptions) error { +func (e *expectedVersionId) applyPut(opts *putOptions) { opts.expectedVersion = &e.versionId - return nil } func (e *expectedVersionId) applyDelete(opts *deleteOptions) { diff --git a/oxia/options_put.go b/oxia/options_put.go index b982d08a3..2452de9b8 100644 --- a/oxia/options_put.go +++ b/oxia/options_put.go @@ -30,14 +30,19 @@ type putOptions struct { // PutOption represents an option for the [SyncClient.Put] operation. type PutOption interface { - applyPut(opts *putOptions) error + applyPut(opts *putOptions) } func newPutOptions(opts []PutOption) (*putOptions, error) { putOpts := &putOptions{} for _, opt := range opts { - if err := opt.applyPut(putOpts); err != nil { - return nil, err + opt.applyPut(putOpts) + } + + for _, secondaryIndex := range putOpts.secondaryIndexes { + if strings.IndexByte(secondaryIndex.indexName, '/') >= 0 { + return nil, errors.Wrapf(ErrInvalidOptions, + "secondary index name %q must not contain '/'", secondaryIndex.indexName) } } @@ -68,9 +73,8 @@ type ephemeral struct{} var ephemeralFlag = &ephemeral{} -func (*ephemeral) applyPut(opts *putOptions) error { +func (*ephemeral) applyPut(opts *putOptions) { opts.ephemeral = true - return nil } // Ephemeral marks the record to be created as an ephemeral record. @@ -89,9 +93,8 @@ type sequenceKeysDeltas struct { sequenceKeysDeltas []uint64 } -func (s *sequenceKeysDeltas) applyPut(opts *putOptions) error { +func (s *sequenceKeysDeltas) applyPut(opts *putOptions) { opts.sequenceKeysDeltas = s.sequenceKeysDeltas - return nil } // SequenceKeysDeltas will request that the final record key to be @@ -110,13 +113,8 @@ type secondaryIdxOption struct { secondaryKey string } -func (s *secondaryIdxOption) applyPut(opts *putOptions) error { - if strings.IndexByte(s.indexName, '/') >= 0 { - return errors.Wrapf(ErrInvalidOptions, - "secondary index name %q must not contain '/'", s.indexName) - } +func (s *secondaryIdxOption) applyPut(opts *putOptions) { opts.secondaryIndexes = append(opts.secondaryIndexes, s) - return nil } // SecondaryIndex lets users specify additional keys to index the record. diff --git a/oxiad/dataserver/controller/lead/secondary_indexes.go b/oxiad/dataserver/controller/lead/secondary_indexes.go index d15cb5ee2..e694f19cb 100644 --- a/oxiad/dataserver/controller/lead/secondary_indexes.go +++ b/oxiad/dataserver/controller/lead/secondary_indexes.go @@ -36,6 +36,10 @@ const secondaryIdxKeyPrefix = constant.InternalKeyPrefix + "idx" type wrapperUpdateCallback struct{} func (wrapperUpdateCallback) ValidatePut(req *proto.PutRequest, features feature.Checker) proto.Status { + if status := sessionManagerUpdateOperationCallback.ValidatePut(req, features); status != proto.Status_OK { + return status + } + return secondaryIndexesUpdateCallback.ValidatePut(req, features) } From 6ffd5100bf6e39f1160b3a9529e4a5a945c56204 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 30 Aug 2026 15:22:25 +0800 Subject: [PATCH 8/8] fix: enable index name validation in standalone Signed-off-by: mattisonchao --- oxiad/dataserver/standalone.go | 4 ++ oxiad/dataserver/standalone_test.go | 59 +++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+) create mode 100644 oxiad/dataserver/standalone_test.go diff --git a/oxiad/dataserver/standalone.go b/oxiad/dataserver/standalone.go index f961422eb..85d012535 100644 --- a/oxiad/dataserver/standalone.go +++ b/oxiad/dataserver/standalone.go @@ -167,6 +167,10 @@ func (s *Standalone) initializeShards(numShards uint32) error { Term: newTerm, ReplicationFactor: 1, FollowerMaps: make(map[string]*proto.EntryId), + // Standalone has no coordinator to negotiate features. Enable the + // locally safe validation feature explicitly without also enabling + // cluster-specific features such as DB checksums. + FeaturesSupported: []proto.Feature{proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION}, }); err != nil { return err } diff --git a/oxiad/dataserver/standalone_test.go b/oxiad/dataserver/standalone_test.go new file mode 100644 index 000000000..a96bd4c43 --- /dev/null +++ b/oxiad/dataserver/standalone_test.go @@ -0,0 +1,59 @@ +// Copyright 2023-2026 The Oxia Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dataserver + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + pb "google.golang.org/protobuf/proto" + + "github.com/oxia-db/oxia/common/proto" +) + +func TestStandaloneSecondaryIndexNameValidation(t *testing.T) { + standaloneServer, err := NewStandalone(NewTestConfig(t.TempDir())) + require.NoError(t, err) + defer standaloneServer.Close() + + leader, err := standaloneServer.shardsDirector.GetLeader(0) + require.NoError(t, err) + require.Eventually(t, func() bool { + return leader.IsFeatureEnabled(proto.Feature_FEATURE_SECONDARY_INDEX_NAME_VALIDATION) + }, 10*time.Second, 10*time.Millisecond) + + conn, err := grpc.NewClient(standaloneServer.ServiceAddr(), grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + defer conn.Close() + + response, err := proto.NewOxiaClientClient(conn).Write(t.Context(), &proto.WriteRequest{ + Shard: pb.Int64(0), + Puts: []*proto.PutRequest{{ + Key: "key", + Value: []byte("value"), + SecondaryIndexes: []*proto.SecondaryIndex{{ + IndexName: "tenant/users", + SecondaryKey: "email", + }}, + }}, + }) + require.NoError(t, err) + require.Len(t, response.GetPuts(), 1) + assert.Equal(t, proto.Status_INVALID_ARGUMENT, response.GetPuts()[0].GetStatus()) +}