From 51de87060fb23543aa290d99fd588ae6fce22416 Mon Sep 17 00:00:00 2001 From: Maksym Dolina Date: Sun, 5 Jul 2026 18:36:13 +0100 Subject: [PATCH 1/4] Fix pagination bug cluster in tagging service (BUG-1, BUG-2, BUG-3) - BUG-1: propagate Cassandra errors from getPopulatedBuckets in GetMembersPaginated as 500 instead of masking them as 404 "tag not found" - BUG-2: a failed bucket read now fails the page (500) instead of silently omitting the bucket's members and advancing the cursor past them; same fix applied to the non-paginated read path (GetTagById / GetMembersNonPaginated) - BUG-3: a cursor pointing past the last populated bucket ends the enumeration instead of wrapping around to bucket 0 and re-delivering the whole tag; invalid/corrupt cursors now return 400 Bad Request instead of silently restarting from the beginning - Remove unused BucketedCursor.TotalCollected field - Add a mock DatabaseClient test seam and service-level tests covering all three bugs plus a two-page pagination regression walk Co-Authored-By: Claude Fable 5 --- taggingapi/tag/tag_member_benchmark_test.go | 8 +- taggingapi/tag/tag_member_service.go | 78 +++++--- .../tag/tag_member_service_bugs_test.go | 189 ++++++++++++++++++ taggingapi/tag/tag_member_service_test.go | 40 ++-- 4 files changed, 265 insertions(+), 50 deletions(-) create mode 100644 taggingapi/tag/tag_member_service_bugs_test.go diff --git a/taggingapi/tag/tag_member_benchmark_test.go b/taggingapi/tag/tag_member_benchmark_test.go index 83f8cb3..a8e24f6 100644 --- a/taggingapi/tag/tag_member_benchmark_test.go +++ b/taggingapi/tag/tag_member_benchmark_test.go @@ -48,7 +48,7 @@ func BenchmarkGenerateBucketedCursor(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { - generateBucketedCursor(i%BucketCount, fmt.Sprintf("member-%d", i), i) + generateBucketedCursor(i%BucketCount, fmt.Sprintf("member-%d", i)) } } @@ -56,13 +56,13 @@ func BenchmarkParseBucketedCursor(b *testing.B) { // Pre-generate cursors cursors := make([]string, 1000) for i := 0; i < 1000; i++ { - cursors[i] = generateBucketedCursor(i%BucketCount, fmt.Sprintf("member-%d", i), i) + cursors[i] = generateBucketedCursor(i%BucketCount, fmt.Sprintf("member-%d", i)) } b.ResetTimer() for i := 0; i < b.N; i++ { - parseBucketedCursor(cursors[i%1000]) + parseBucketedCursor(cursors[i%1000]) //nolint:errcheck // benchmark only measures parse cost } } @@ -202,7 +202,7 @@ func BenchmarkGetMembersV2PaginatedWithCursor(b *testing.B) { } tagId := "benchmark-tag-pagination-cursor" - cursor := generateBucketedCursor(100, "member-12345", 500) + cursor := generateBucketedCursor(100, "member-12345") b.ResetTimer() for i := 0; i < b.N; i++ { diff --git a/taggingapi/tag/tag_member_service.go b/taggingapi/tag/tag_member_service.go index 57eb48f..c72eb1d 100644 --- a/taggingapi/tag/tag_member_service.go +++ b/taggingapi/tag/tag_member_service.go @@ -45,12 +45,13 @@ const ( QueryDeleteBucketMetadata = `DELETE FROM tag_member_metadata WHERE tenant_id = ? AND shard_id = ? AND tag_id = ? AND bucket_id = ?` CountMembersCassandraResp = "count" + + InvalidCursorErrorMsg = "invalid pagination cursor" ) type BucketedCursor struct { - BucketId int `json:"bucketId"` - LastMember string `json:"lastMember,omitempty"` - TotalCollected int `json:"totalCollected"` + BucketId int `json:"bucketId"` + LastMember string `json:"lastMember,omitempty"` } type PaginatedMembersResponse struct { @@ -244,9 +245,15 @@ func GetMembersPaginated(tenantId string, tagId string, limit int, cursor string log.Debugf("Getting paginated members for tag %s, limit %d, cursor %s", tagId, limit, cursor) + state, err := parseBucketedCursor(cursor) + if err != nil { + return nil, err + } + populatedBuckets, err := getPopulatedBuckets(tenantId, tagId) if err != nil { log.Errorf("Error getting populated buckets for tag %s: %v", tagId, err) + return nil, fmt.Errorf("failed to get populated buckets: %w", err) } if len(populatedBuckets) == 0 { @@ -255,16 +262,24 @@ func GetMembersPaginated(tenantId string, tagId string, limit int, cursor string log.Debugf("Found %d populated buckets for tag %s", len(populatedBuckets), tagId) - state := parseBucketedCursor(cursor) var allMembers []string - startIndex := 0 + // Find the first populated bucket at or after the cursor position. If none + // exists (buckets were emptied since the previous page), the enumeration is + // complete — do not wrap around to the beginning. + startIndex := -1 for i, bucketId := range populatedBuckets { if bucketId >= state.BucketId { startIndex = i break } } + if startIndex == -1 { + return &PaginatedMembersResponse{ + Data: []string{}, + HasMore: false, + }, nil + } // Build work items for remaining buckets (apply cursor's lastMember to first bucket only) workers := getReadWorkerCount() @@ -285,10 +300,16 @@ func GetMembersPaginated(tenantId string, tagId string, limit int, cursor string orderedResults := fetchBucketsConcurrent(tenantId, tagId, workItems, workers) - // Merge in bucket order, building cursor at the truncation point + // Merge in bucket order, building cursor at the truncation point. + // A failed bucket fails the whole page: returning partial data would let the + // cursor advance past the failed bucket and silently omit its members from + // the enumeration. The caller retries with the same cursor instead. lastProcessedBucketIndex := startIndex - 1 for idx, result := range orderedResults { - if result.err != nil || len(result.members) == 0 { + if result.err != nil { + return nil, fmt.Errorf("failed to fetch members from bucket %d: %w", remainingBuckets[idx], result.err) + } + if len(result.members) == 0 { lastProcessedBucketIndex = startIndex + idx continue } @@ -298,7 +319,7 @@ func GetMembersPaginated(tenantId string, tagId string, limit int, cursor string if len(result.members) > needed { allMembers = append(allMembers, result.members[:needed]...) - nextCursor := generateBucketedCursor(currentBucketId, result.members[needed-1], len(allMembers)) + nextCursor := generateBucketedCursor(currentBucketId, result.members[needed-1]) log.Debugf("Returning %d members for tag %s with more data in bucket %d", len(allMembers), tagId, currentBucketId) return &PaginatedMembersResponse{ @@ -321,7 +342,7 @@ func GetMembersPaginated(tenantId string, tagId string, limit int, cursor string var nextCursor string if hasMore { nextBucketId := populatedBuckets[lastProcessedBucketIndex+1] - nextCursor = generateBucketedCursor(nextBucketId, "", 0) + nextCursor = generateBucketedCursor(nextBucketId, "") } log.Debugf("Returning %d members for tag %s, hasMore: %v", len(allMembers), tagId, hasMore) @@ -360,11 +381,10 @@ func getMembersFromBucket(tenantId string, tagId string, bucketId int, lastMembe } // Cursor management functions -func generateBucketedCursor(bucketId int, lastMember string, totalCollected int) string { +func generateBucketedCursor(bucketId int, lastMember string) string { cursor := BucketedCursor{ - BucketId: bucketId, - LastMember: lastMember, - TotalCollected: totalCollected, + BucketId: bucketId, + LastMember: lastMember, } data, err := json.Marshal(cursor) @@ -375,30 +395,26 @@ func generateBucketedCursor(bucketId int, lastMember string, totalCollected int) return base64.URLEncoding.EncodeToString(data) } -func parseBucketedCursor(cursor string) BucketedCursor { +func parseBucketedCursor(cursor string) (BucketedCursor, error) { if cursor == "" { - return BucketedCursor{BucketId: 0} + return BucketedCursor{BucketId: 0}, nil } data, err := base64.URLEncoding.DecodeString(cursor) if err != nil { - log.Errorf("Error decoding cursor: %v", err) - return BucketedCursor{BucketId: 0} + return BucketedCursor{}, xwcommon.NewRemoteErrorAS(http.StatusBadRequest, InvalidCursorErrorMsg) } var state BucketedCursor if err := json.Unmarshal(data, &state); err != nil { - log.Errorf("Error unmarshaling cursor: %v", err) - return BucketedCursor{BucketId: 0} + return BucketedCursor{}, xwcommon.NewRemoteErrorAS(http.StatusBadRequest, InvalidCursorErrorMsg) } - // Validate cursor values if state.BucketId < 0 || state.BucketId >= BucketCount { - log.Warnf("Invalid bucket ID in cursor: %d, resetting to 0", state.BucketId) - return BucketedCursor{BucketId: 0} + return BucketedCursor{}, xwcommon.NewRemoteErrorAS(http.StatusBadRequest, InvalidCursorErrorMsg) } - return state + return state, nil } func min(a, b int) int { @@ -532,16 +548,22 @@ func fetchMembersFromBucketsConcurrent(tenantId string, tagId string, bucketIds orderedResults := fetchBucketsConcurrent(tenantId, tagId, workItems, workers) - // Merge in bucket order, stop at totalLimit + // Merge in bucket order, stop at totalLimit. + // A failed bucket fails the whole read — a partial result would silently + // under-report the tag's membership. collected := make([]string, 0) - for _, result := range orderedResults { - if result.err != nil || len(result.members) == 0 { - continue - } + for idx, result := range orderedResults { space := totalLimit - len(collected) if space <= 0 { + // Response already full — errors in later buckets are irrelevant return collected, true, nil } + if result.err != nil { + return nil, false, fmt.Errorf("failed to fetch members from bucket %d: %w", bucketIds[idx], result.err) + } + if len(result.members) == 0 { + continue + } if len(result.members) > space { collected = append(collected, result.members[:space]...) return collected, true, nil diff --git a/taggingapi/tag/tag_member_service_bugs_test.go b/taggingapi/tag/tag_member_service_bugs_test.go new file mode 100644 index 0000000..6a966a4 --- /dev/null +++ b/taggingapi/tag/tag_member_service_bugs_test.go @@ -0,0 +1,189 @@ +package tag + +import ( + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/rdkcentral/xconfadmin/common" + + xwcommon "github.com/rdkcentral/xconfwebconfig/common" + "github.com/rdkcentral/xconfwebconfig/db" + + "github.com/gorilla/mux" + "github.com/stretchr/testify/assert" +) + +// mockDbClient injects Cassandra query behavior for service-level tests. +// The embedded interface panics on any method the test does not stub, +// which flags unexpected database usage. +type mockDbClient struct { + db.DatabaseClient + queryFunc func(query string, params ...string) ([]map[string]any, error) +} + +func (m *mockDbClient) QueryXconfDataRows(query string, params ...string) ([]map[string]any, error) { + return m.queryFunc(query, params...) +} + +func withMockDb(t *testing.T, queryFunc func(query string, params ...string) ([]map[string]any, error)) { + t.Helper() + old := db.GetDatabaseClient() + db.SetDatabaseClient(&mockDbClient{queryFunc: queryFunc}) + t.Cleanup(func() { db.SetDatabaseClient(old) }) +} + +func isMetadataQuery(query string) bool { + return strings.Contains(query, "tag_member_metadata") +} + +// BUG-1: a Cassandra failure while listing populated buckets must surface as an +// internal error, not as 404 "tag not found". +func TestGetMembersPaginated_DbErrorIsNot404(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + return nil, errors.New("cassandra unavailable") + }) + + resp, err := GetMembersPaginated(db.GetDefaultTenantId(), "some-tag", 10, "") + assert.Nil(t, resp) + assert.Error(t, err) + assert.Equal(t, http.StatusInternalServerError, xwcommon.GetXconfErrorStatusCode(err)) +} + +// Unknown tag (no populated buckets, no DB error) is still a 404. +func TestGetMembersPaginated_UnknownTagIs404(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + return []map[string]any{}, nil + }) + + resp, err := GetMembersPaginated(db.GetDefaultTenantId(), "missing-tag", 10, "") + assert.Nil(t, resp) + assert.Error(t, err) + assert.Equal(t, http.StatusNotFound, xwcommon.GetXconfErrorStatusCode(err)) +} + +// BUG-2: a failed bucket read fails the page instead of silently omitting that +// bucket's members and advancing the cursor past them. +func TestGetMembersPaginated_BucketFetchErrorFailsPage(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}, {"bucket_id": 2}}, nil + } + if params[2] == "2" { + return nil, errors.New("read timeout") + } + return []map[string]any{{"member": "AA:BB:CC:DD:EE:01"}}, nil + }) + + resp, err := GetMembersPaginated(db.GetDefaultTenantId(), "some-tag", 10, "") + assert.Nil(t, resp) + assert.Error(t, err) + assert.Contains(t, err.Error(), "bucket 2") + assert.Equal(t, http.StatusInternalServerError, xwcommon.GetXconfErrorStatusCode(err)) +} + +// BUG-2 (non-paginated path): GetTagById must not return a silently incomplete +// member list when a bucket read fails. +func TestGetTagById_BucketFetchErrorFails(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 7}}, nil + } + return nil, errors.New("read timeout") + }) + + members, truncated, err := GetTagById(db.GetDefaultTenantId(), "some-tag") + assert.Error(t, err) + assert.Nil(t, members) + assert.False(t, truncated) +} + +// BUG-3: a cursor pointing past the last populated bucket ends the enumeration +// instead of wrapping around to bucket 0 and re-delivering everything. +func TestGetMembersPaginated_CursorBeyondLastBucketEndsEnumeration(t *testing.T) { + setupTestEnvironment() + var memberQueries atomic.Int32 + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}, {"bucket_id": 2}}, nil + } + memberQueries.Add(1) + return []map[string]any{{"member": "m1"}}, nil + }) + + cursor := generateBucketedCursor(500, "") + resp, err := GetMembersPaginated(db.GetDefaultTenantId(), "some-tag", 10, cursor) + assert.NoError(t, err) + assert.NotNil(t, resp) + assert.Empty(t, resp.Data) + assert.False(t, resp.HasMore) + assert.Empty(t, resp.NextCursor) + assert.Equal(t, int32(0), memberQueries.Load(), "no member fetches expected past the last populated bucket") +} + +// BUG-3 (related): an unparseable cursor is a 400, not a silent restart from +// bucket 0. +func TestGetMembersPaginated_InvalidCursorReturns400(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + t.Error("no database query expected for an invalid cursor") + return nil, nil + }) + + resp, err := GetMembersPaginated(db.GetDefaultTenantId(), "some-tag", 10, "!!!not-a-cursor!!!") + assert.Nil(t, resp) + assert.Error(t, err) + assert.Equal(t, http.StatusBadRequest, xwcommon.GetXconfErrorStatusCode(err)) +} + +func TestGetTagMembersHandler_InvalidCursorReturns400(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + return nil, errors.New("unexpected db call") + }) + + req := httptest.NewRequest("GET", "/taggingService/tags/some-tag/members?cursor=%21bad%21", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "some-tag"}) + w := httptest.NewRecorder() + GetTagMembersHandler(w, req) + + assert.Equal(t, http.StatusBadRequest, w.Code) + assert.Contains(t, w.Body.String(), InvalidCursorErrorMsg) +} + +// Regression: two-page walk over two buckets returns each member exactly once +// and terminates. +func TestGetMembersPaginated_TwoPageWalk(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}, {"bucket_id": 2}}, nil + } + switch params[2] { + case "1": + return []map[string]any{{"member": "m1"}}, nil + case "2": + return []map[string]any{{"member": "m2"}}, nil + } + return nil, fmt.Errorf("unexpected bucket %s", params[2]) + }) + + page1, err := GetMembersPaginated(db.GetDefaultTenantId(), "some-tag", 1, "") + assert.NoError(t, err) + assert.Equal(t, []string{"m1"}, page1.Data) + assert.True(t, page1.HasMore) + assert.NotEmpty(t, page1.NextCursor) + + page2, err := GetMembersPaginated(db.GetDefaultTenantId(), "some-tag", 1, page1.NextCursor) + assert.NoError(t, err) + assert.Equal(t, []string{"m2"}, page2.Data) + assert.False(t, page2.HasMore) +} diff --git a/taggingapi/tag/tag_member_service_test.go b/taggingapi/tag/tag_member_service_test.go index 0482bf6..4bdba07 100644 --- a/taggingapi/tag/tag_member_service_test.go +++ b/taggingapi/tag/tag_member_service_test.go @@ -2,8 +2,10 @@ package tag import ( "fmt" + "net/http" "testing" + xwcommon "github.com/rdkcentral/xconfwebconfig/common" "github.com/rdkcentral/xconfwebconfig/db" "github.com/stretchr/testify/assert" ) @@ -43,46 +45,48 @@ func TestGetBucketId(t *testing.T) { func TestParseBucketedCursor(t *testing.T) { // Test empty cursor - cursor := parseBucketedCursor("") + cursor, err := parseBucketedCursor("") + assert.NoError(t, err) assert.Equal(t, 0, cursor.BucketId) assert.Equal(t, "", cursor.LastMember) - assert.Equal(t, 0, cursor.TotalCollected) // Test valid cursor - validCursor := generateBucketedCursor(5, "test-member", 100) - parsed := parseBucketedCursor(validCursor) + validCursor := generateBucketedCursor(5, "test-member") + parsed, err := parseBucketedCursor(validCursor) + assert.NoError(t, err) assert.Equal(t, 5, parsed.BucketId) assert.Equal(t, "test-member", parsed.LastMember) - assert.Equal(t, 100, parsed.TotalCollected) - // Test invalid cursor - invalidCursor := parseBucketedCursor("invalid-cursor") - assert.Equal(t, 0, invalidCursor.BucketId) + // Invalid cursor is a 400 error, not a silent reset to bucket 0 + _, err = parseBucketedCursor("invalid-cursor") + assert.Error(t, err) + assert.Equal(t, http.StatusBadRequest, xwcommon.GetXconfErrorStatusCode(err)) - // Test cursor with invalid bucket ID - invalidBucketCursor := generateBucketedCursor(9999, "member", 100) - parsed2 := parseBucketedCursor(invalidBucketCursor) - assert.Equal(t, 0, parsed2.BucketId, "Invalid bucket ID should be reset to 0") + // Out-of-range bucket ID is a 400 error + invalidBucketCursor := generateBucketedCursor(9999, "member") + _, err = parseBucketedCursor(invalidBucketCursor) + assert.Error(t, err) + assert.Equal(t, http.StatusBadRequest, xwcommon.GetXconfErrorStatusCode(err)) } func TestGenerateBucketedCursor(t *testing.T) { - cursor := generateBucketedCursor(10, "member123", 500) + cursor := generateBucketedCursor(10, "member123") assert.NotEmpty(t, cursor, "Cursor should not be empty") // Should be base64 encoded - parsed := parseBucketedCursor(cursor) + parsed, err := parseBucketedCursor(cursor) + assert.NoError(t, err) assert.Equal(t, 10, parsed.BucketId) assert.Equal(t, "member123", parsed.LastMember) - assert.Equal(t, 500, parsed.TotalCollected) // Test edge cases - cursor2 := generateBucketedCursor(0, "", 0) + cursor2 := generateBucketedCursor(0, "") assert.NotEmpty(t, cursor2, "Cursor should not be empty even with zero values") - parsed2 := parseBucketedCursor(cursor2) + parsed2, err := parseBucketedCursor(cursor2) + assert.NoError(t, err) assert.Equal(t, 0, parsed2.BucketId) assert.Equal(t, "", parsed2.LastMember) - assert.Equal(t, 0, parsed2.TotalCollected) } func TestBucketDistribution(t *testing.T) { From 55146d3b0a912ca9b902de5f7e98f544861b3e83 Mon Sep 17 00:00:00 2001 From: Maksym Dolina Date: Sun, 5 Jul 2026 18:49:45 +0100 Subject: [PATCH 2/4] Fix tag deletion bug cluster (BUG-4, BUG-5, BUG-6, BUG-7 minors) - BUG-4: deleteBucketMembers returns an error on partial XDAS removal instead of nil, so DeleteTag no longer reports a half-deleted tag as a completed deletion; bucket metadata survives for retry - BUG-5: per-instance in-flight guard in DeleteTagHandler deduplicates concurrent background deletions of the same tag (interim until the tag registry provides cross-instance state) - BUG-6: GetTagById returns a typed 404 error; handler no longer matches the error message by string comparison - BUG-7: remove min/max helpers shadowing Go builtins; skip the COUNT(*) metadata-cleanup check for buckets whose batch delete failed - Add exported template setters to GroupServiceSyncConnector so tests can point it at a mock XDAS server; extend the mock DatabaseClient with batch/modify support Co-Authored-By: Claude Fable 5 --- http/groupsync_service_connector.go | 8 + taggingapi/tag/tag_delete_bugs_test.go | 175 ++++++++++++++++++ taggingapi/tag/tag_member_handler.go | 28 ++- taggingapi/tag/tag_member_service.go | 33 ++-- .../tag/tag_member_service_bugs_test.go | 46 ++++- 5 files changed, 260 insertions(+), 30 deletions(-) create mode 100644 taggingapi/tag/tag_delete_bugs_test.go diff --git a/http/groupsync_service_connector.go b/http/groupsync_service_connector.go index d0f2024..2abc58f 100644 --- a/http/groupsync_service_connector.go +++ b/http/groupsync_service_connector.go @@ -70,6 +70,14 @@ func (c *GroupServiceSyncConnector) SetGroupServiceSyncHost(host string) { c.BaseURL = host } +func (c *GroupServiceSyncConnector) SetAddGroupMemberTemplate(template string) { + c.addGroupMemberTemplate = template +} + +func (c *GroupServiceSyncConnector) SetRemoveGroupMemberTemplate(template string) { + c.removeGroupMemberTemplate = template +} + func (c *GroupServiceSyncConnector) DoRequest(method string, url string, headers map[string]string, body []byte) ([]byte, error) { rbytes, err := c.Client.DoWithRetries(method, url, headers, body, log.Fields{}, groupServiceSyncServiceName) return rbytes, err diff --git a/taggingapi/tag/tag_delete_bugs_test.go b/taggingapi/tag/tag_delete_bugs_test.go new file mode 100644 index 0000000..2101a65 --- /dev/null +++ b/taggingapi/tag/tag_delete_bugs_test.go @@ -0,0 +1,175 @@ +package tag + +import ( + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/rdkcentral/xconfadmin/common" + xhttp "github.com/rdkcentral/xconfadmin/http" + + "github.com/rdkcentral/xconfwebconfig/db" + xwhttp "github.com/rdkcentral/xconfwebconfig/http" + + "github.com/gorilla/mux" + "github.com/stretchr/testify/assert" +) + +// withMockXdasSync points the GroupServiceSyncConnector at an httptest server +// for the duration of the test. +func withMockXdasSync(t *testing.T, handler http.HandlerFunc) { + t.Helper() + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + + connector := xhttp.WebConfServer.GroupServiceSyncConnector + oldHost := connector.GetGroupServiceSyncHost() + oldClient := connector.Client + connector.SetGroupServiceSyncHost(server.URL) + connector.SetAddGroupMemberTemplate("%s/ft/%s") + connector.SetRemoveGroupMemberTemplate("%s/ft/%s?field=%s") + connector.Client = &xhttp.HttpClient{Client: server.Client()} + t.Cleanup(func() { + connector.SetGroupServiceSyncHost(oldHost) + connector.Client = oldClient + }) +} + +// BUG-4: when XDAS removes only part of a bucket's members, DeleteTag must +// report an error instead of claiming a completed deletion. +func TestDeleteTag_PartialXdasFailureReturnsError(t *testing.T) { + setupTestEnvironment() + withMockXdasSync(t, func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "MFAIL") { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusOK) + }) + + var executedBatches []string + mock := &mockDbClient{ + queryFunc: func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}}, nil + } + if strings.Contains(query, "count(*)") { + return []map[string]any{{"count": int64(1)}}, nil + } + return []map[string]any{{"member": "MFAIL"}, {"member": "MOK"}}, nil + }, + execBatchFn: func(batch *mockBatch) error { + executedBatches = append(executedBatches, batch.statements...) + return nil + }, + } + withMockDbClient(t, mock) + + err := DeleteTag(db.GetDefaultTenantId(), "some-tag") + assert.Error(t, err) + assert.Contains(t, err.Error(), "partial XDAS deletion") + + // Bucket metadata must survive so a retry can resume the deletion + for _, stmt := range executedBatches { + assert.NotContains(t, stmt, "tag_member_metadata", + "bucket metadata must not be deleted after a partial XDAS failure") + } +} + +// BUG-5: a second DELETE for a tag whose deletion is already running returns +// 202 without spawning another background deleter. +func TestDeleteTagHandler_DeduplicatesConcurrentDeletions(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}}, nil + } + t.Error("no member fetches expected while deletion is already in flight") + return nil, nil + }) + + deletionKey := db.GetDefaultTenantId() + "|dup-tag" + inFlightTagDeletions.Store(deletionKey, true) + t.Cleanup(func() { inFlightTagDeletions.Delete(deletionKey) }) + + req := httptest.NewRequest("DELETE", "/taggingService/tags/dup-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "dup-tag"}) + rec := httptest.NewRecorder() + DeleteTagHandler(xwhttp.NewXResponseWriter(rec), req) + + assert.Equal(t, http.StatusAccepted, rec.Code) + assert.Contains(t, rec.Body.String(), "already in progress") +} + +// BUG-5: the guard is released once the background deletion finishes, so a +// later DELETE for the same tag is accepted again. +func TestDeleteTagHandler_ReleasesGuardAfterCompletion(t *testing.T) { + setupTestEnvironment() + withMockXdasSync(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}}, nil + } + // Empty bucket: deletion completes immediately + return []map[string]any{}, nil + }) + + req := httptest.NewRequest("DELETE", "/taggingService/tags/guard-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "guard-tag"}) + rec := httptest.NewRecorder() + DeleteTagHandler(xwhttp.NewXResponseWriter(rec), req) + assert.Equal(t, http.StatusAccepted, rec.Code) + assert.Contains(t, rec.Body.String(), "queued for processing") + + // Wait for the background deletion to finish and release the guard. + // This must complete before the test ends so the mock DB is not restored + // under a still-running goroutine. + deletionKey := db.GetDefaultTenantId() + "|guard-tag" + deadline := time.Now().Add(5 * time.Second) + for { + if _, running := inFlightTagDeletions.Load(deletionKey); !running { + break + } + if time.Now().After(deadline) { + t.Fatal("background deletion did not release the in-flight guard") + } + time.Sleep(10 * time.Millisecond) + } +} + +// BUG-6: unknown tag on GET /tags/{tag} is a typed 404, not a string-matched +// error. +func TestGetTagByIdHandler_UnknownTagReturns404(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + return []map[string]any{}, nil + }) + + req := httptest.NewRequest("GET", "/taggingService/tags/missing-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "missing-tag"}) + w := httptest.NewRecorder() + GetTagByIdHandler(w, req) + + assert.Equal(t, http.StatusNotFound, w.Code) + assert.Contains(t, w.Body.String(), "missing-tag tag not found") +} + +// BUG-6: a database failure on GET /tags/{tag} is a 500, not a 404. +func TestGetTagByIdHandler_DbErrorReturns500(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + return nil, errors.New("cassandra unavailable") + }) + + req := httptest.NewRequest("GET", "/taggingService/tags/some-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "some-tag"}) + w := httptest.NewRecorder() + GetTagByIdHandler(w, req) + + assert.Equal(t, http.StatusInternalServerError, w.Code) +} diff --git a/taggingapi/tag/tag_member_handler.go b/taggingapi/tag/tag_member_handler.go index 0ab739c..acc3a40 100644 --- a/taggingapi/tag/tag_member_handler.go +++ b/taggingapi/tag/tag_member_handler.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "strconv" + "sync" "github.com/rdkcentral/xconfadmin/common" xhttp "github.com/rdkcentral/xconfadmin/http" @@ -272,11 +273,6 @@ func GetTagByIdHandler(w http.ResponseWriter, r *http.Request) { tenantId := xwhttp.GetTenantId(r, "") members, wasTruncated, err := GetTagById(tenantId, id) if err != nil { - // Check if tag not found - if err.Error() == "tag not found" { - xhttp.WriteXconfResponse(w, http.StatusNotFound, []byte(fmt.Sprintf(NotFoundErrorMsg, id))) - return - } xhttp.WriteXconfErrorResponse(w, err) return } @@ -305,6 +301,11 @@ func GetTagByIdHandler(w http.ResponseWriter, r *http.Request) { xhttp.WriteXconfResponse(w, statusCode, respBytes) } +// inFlightTagDeletions deduplicates concurrent background deletions of the +// same tag on this instance, keyed by "tenantId|tagId". Interim guard until +// tag deletion state is tracked cross-instance (tag registry, topic 3). +var inFlightTagDeletions sync.Map + // DeleteTagHandler deletes a tag and all its members from V2 storage asynchronously func DeleteTagHandler(w http.ResponseWriter, r *http.Request) { id, found := mux.Vars(r)[common.Tag] @@ -331,8 +332,25 @@ func DeleteTagHandler(w http.ResponseWriter, r *http.Request) { return } + deletionKey := tenantId + "|" + id + if _, alreadyRunning := inFlightTagDeletions.LoadOrStore(deletionKey, true); alreadyRunning { + response := map[string]string{ + "status": "accepted", + "message": fmt.Sprintf("Tag '%s' deletion is already in progress", id), + "tag": id, + } + respBytes, err := json.Marshal(response) + if err != nil { + xhttp.WriteXconfErrorResponse(w, err) + return + } + xhttp.WriteXconfResponse(w, http.StatusAccepted, respBytes) + return + } + auditId := xw.AuditId() go func(tagId string) { + defer inFlightTagDeletions.Delete(deletionKey) if err := DeleteTag(tenantId, tagId); err != nil { log.WithFields(log.Fields{ "audit_id": auditId, diff --git a/taggingapi/tag/tag_member_service.go b/taggingapi/tag/tag_member_service.go index c72eb1d..06d2d42 100644 --- a/taggingapi/tag/tag_member_service.go +++ b/taggingapi/tag/tag_member_service.go @@ -158,11 +158,13 @@ func RemoveMembers(tenantId string, tagId string, members []string) error { allErrors = append(allErrors, fmt.Sprintf("bucket %d: %v", bucketId, err)) log.Errorf("Failed to remove %d members from bucket %d for tag %s: %v", len(bucketMembers), bucketId, tagId, err) - } else { - successCount += len(bucketMembers) - log.Debugf("Successfully removed %d members from bucket %d for tag %s", - len(bucketMembers), bucketId, tagId) + // The delete failed, so the bucket cannot have become empty — + // skip the metadata cleanup check + continue } + successCount += len(bucketMembers) + log.Debugf("Successfully removed %d members from bucket %d for tag %s", + len(bucketMembers), bucketId, tagId) // Clean up bucket metadata if bucket is now empty membersCount, err := getMembersCountOfBucket(tenantId, tagId, bucketId) if err != nil { @@ -417,20 +419,6 @@ func parseBucketedCursor(cursor string) (BucketedCursor, error) { return state, nil } -func min(a, b int) int { - if a < b { - return a - } - return b -} - -func max(a, b int) int { - if a > b { - return a - } - return b -} - // getReadWorkerCount returns the worker count for concurrent read operations func getReadWorkerCount() int { config := GetTagApiConfig() @@ -776,7 +764,7 @@ func GetTagById(tenantId string, tagId string) ([]string, bool, error) { } if len(populatedBuckets) == 0 { - return nil, false, fmt.Errorf("tag not found") + return nil, false, xwcommon.NewRemoteErrorAS(http.StatusNotFound, fmt.Sprintf(NotFoundErrorMsg, tagId)) } log.Infof("Fetching tag '%s' with %d populated buckets", tagId, len(populatedBuckets)) @@ -866,8 +854,11 @@ func deleteBucketMembers(tenantId string, tagId string, bucketId int) (int, erro } if len(removedFromXdas) < len(chunk) { - log.Warnf("partial XDAS deletion: %d/%d members removed", len(removedFromXdas), len(chunk)) - return totalDeleted, nil + // Partial XDAS failure must fail the bucket: returning success here + // would let DeleteTag report a completed deletion while leftover + // members and bucket metadata remain in both stores. + return totalDeleted, fmt.Errorf("partial XDAS deletion in bucket %d: %d/%d members removed", + bucketId, len(removedFromXdas), len(chunk)) } if len(chunk) < MaxBatchSizeV2 { diff --git a/taggingapi/tag/tag_member_service_bugs_test.go b/taggingapi/tag/tag_member_service_bugs_test.go index 6a966a4..21d4a74 100644 --- a/taggingapi/tag/tag_member_service_bugs_test.go +++ b/taggingapi/tag/tag_member_service_bugs_test.go @@ -20,23 +20,61 @@ import ( // mockDbClient injects Cassandra query behavior for service-level tests. // The embedded interface panics on any method the test does not stub, -// which flags unexpected database usage. +// which flags unexpected database usage. Batch and modify operations succeed +// by default; set the optional funcs to override. type mockDbClient struct { db.DatabaseClient - queryFunc func(query string, params ...string) ([]map[string]any, error) + queryFunc func(query string, params ...string) ([]map[string]any, error) + modifyFunc func(query string, params ...string) error + execBatchFn func(batch *mockBatch) error +} + +type mockBatch struct { + statements []string +} + +func (b *mockBatch) Query(stmt string, args ...any) { + b.statements = append(b.statements, stmt) +} + +func (b *mockBatch) Size() int { + return len(b.statements) } func (m *mockDbClient) QueryXconfDataRows(query string, params ...string) ([]map[string]any, error) { return m.queryFunc(query, params...) } -func withMockDb(t *testing.T, queryFunc func(query string, params ...string) ([]map[string]any, error)) { +func (m *mockDbClient) ModifyXconfData(query string, params ...string) error { + if m.modifyFunc != nil { + return m.modifyFunc(query, params...) + } + return nil +} + +func (m *mockDbClient) NewBatch(batchType int) db.BatchOperation { + return &mockBatch{} +} + +func (m *mockDbClient) ExecuteBatch(batch db.BatchOperation) error { + if m.execBatchFn != nil { + return m.execBatchFn(batch.(*mockBatch)) + } + return nil +} + +func withMockDbClient(t *testing.T, client *mockDbClient) { t.Helper() old := db.GetDatabaseClient() - db.SetDatabaseClient(&mockDbClient{queryFunc: queryFunc}) + db.SetDatabaseClient(client) t.Cleanup(func() { db.SetDatabaseClient(old) }) } +func withMockDb(t *testing.T, queryFunc func(query string, params ...string) ([]map[string]any, error)) { + t.Helper() + withMockDbClient(t, &mockDbClient{queryFunc: queryFunc}) +} + func isMetadataQuery(query string) bool { return strings.Contains(query, "tag_member_metadata") } From 4bdc85d77410633ab14feca31defd1052039acc0 Mon Sep 17 00:00:00 2001 From: Maksym Dolina Date: Sun, 5 Jul 2026 18:49:45 +0100 Subject: [PATCH 3/4] Fix tag deletion bug cluster (BUG-4, BUG-5, BUG-6, BUG-7 minors) - BUG-4: deleteBucketMembers returns an error on partial XDAS removal instead of nil, so DeleteTag no longer reports a half-deleted tag as a completed deletion; bucket metadata survives for retry - BUG-5: per-instance in-flight guard in DeleteTagHandler deduplicates concurrent background deletions of the same tag (interim until the tag registry provides cross-instance state) - BUG-6: GetTagById returns a typed 404 error; handler no longer matches the error message by string comparison - BUG-7: remove min/max helpers shadowing Go builtins; skip the COUNT(*) metadata-cleanup check for buckets whose batch delete failed - Add exported template setters to GroupServiceSyncConnector so tests can point it at a mock XDAS server; extend the mock DatabaseClient with batch/modify support Co-Authored-By: Claude Fable 5 --- http/groupsync_service_connector.go | 8 + taggingapi/tag/tag_delete_bugs_test.go | 175 ++++++++++++++++++ taggingapi/tag/tag_member_handler.go | 28 ++- taggingapi/tag/tag_member_service.go | 33 ++-- .../tag/tag_member_service_bugs_test.go | 46 ++++- 5 files changed, 260 insertions(+), 30 deletions(-) create mode 100644 taggingapi/tag/tag_delete_bugs_test.go diff --git a/http/groupsync_service_connector.go b/http/groupsync_service_connector.go index d0f2024..2abc58f 100644 --- a/http/groupsync_service_connector.go +++ b/http/groupsync_service_connector.go @@ -70,6 +70,14 @@ func (c *GroupServiceSyncConnector) SetGroupServiceSyncHost(host string) { c.BaseURL = host } +func (c *GroupServiceSyncConnector) SetAddGroupMemberTemplate(template string) { + c.addGroupMemberTemplate = template +} + +func (c *GroupServiceSyncConnector) SetRemoveGroupMemberTemplate(template string) { + c.removeGroupMemberTemplate = template +} + func (c *GroupServiceSyncConnector) DoRequest(method string, url string, headers map[string]string, body []byte) ([]byte, error) { rbytes, err := c.Client.DoWithRetries(method, url, headers, body, log.Fields{}, groupServiceSyncServiceName) return rbytes, err diff --git a/taggingapi/tag/tag_delete_bugs_test.go b/taggingapi/tag/tag_delete_bugs_test.go new file mode 100644 index 0000000..2101a65 --- /dev/null +++ b/taggingapi/tag/tag_delete_bugs_test.go @@ -0,0 +1,175 @@ +package tag + +import ( + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/rdkcentral/xconfadmin/common" + xhttp "github.com/rdkcentral/xconfadmin/http" + + "github.com/rdkcentral/xconfwebconfig/db" + xwhttp "github.com/rdkcentral/xconfwebconfig/http" + + "github.com/gorilla/mux" + "github.com/stretchr/testify/assert" +) + +// withMockXdasSync points the GroupServiceSyncConnector at an httptest server +// for the duration of the test. +func withMockXdasSync(t *testing.T, handler http.HandlerFunc) { + t.Helper() + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + + connector := xhttp.WebConfServer.GroupServiceSyncConnector + oldHost := connector.GetGroupServiceSyncHost() + oldClient := connector.Client + connector.SetGroupServiceSyncHost(server.URL) + connector.SetAddGroupMemberTemplate("%s/ft/%s") + connector.SetRemoveGroupMemberTemplate("%s/ft/%s?field=%s") + connector.Client = &xhttp.HttpClient{Client: server.Client()} + t.Cleanup(func() { + connector.SetGroupServiceSyncHost(oldHost) + connector.Client = oldClient + }) +} + +// BUG-4: when XDAS removes only part of a bucket's members, DeleteTag must +// report an error instead of claiming a completed deletion. +func TestDeleteTag_PartialXdasFailureReturnsError(t *testing.T) { + setupTestEnvironment() + withMockXdasSync(t, func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "MFAIL") { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusOK) + }) + + var executedBatches []string + mock := &mockDbClient{ + queryFunc: func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}}, nil + } + if strings.Contains(query, "count(*)") { + return []map[string]any{{"count": int64(1)}}, nil + } + return []map[string]any{{"member": "MFAIL"}, {"member": "MOK"}}, nil + }, + execBatchFn: func(batch *mockBatch) error { + executedBatches = append(executedBatches, batch.statements...) + return nil + }, + } + withMockDbClient(t, mock) + + err := DeleteTag(db.GetDefaultTenantId(), "some-tag") + assert.Error(t, err) + assert.Contains(t, err.Error(), "partial XDAS deletion") + + // Bucket metadata must survive so a retry can resume the deletion + for _, stmt := range executedBatches { + assert.NotContains(t, stmt, "tag_member_metadata", + "bucket metadata must not be deleted after a partial XDAS failure") + } +} + +// BUG-5: a second DELETE for a tag whose deletion is already running returns +// 202 without spawning another background deleter. +func TestDeleteTagHandler_DeduplicatesConcurrentDeletions(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}}, nil + } + t.Error("no member fetches expected while deletion is already in flight") + return nil, nil + }) + + deletionKey := db.GetDefaultTenantId() + "|dup-tag" + inFlightTagDeletions.Store(deletionKey, true) + t.Cleanup(func() { inFlightTagDeletions.Delete(deletionKey) }) + + req := httptest.NewRequest("DELETE", "/taggingService/tags/dup-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "dup-tag"}) + rec := httptest.NewRecorder() + DeleteTagHandler(xwhttp.NewXResponseWriter(rec), req) + + assert.Equal(t, http.StatusAccepted, rec.Code) + assert.Contains(t, rec.Body.String(), "already in progress") +} + +// BUG-5: the guard is released once the background deletion finishes, so a +// later DELETE for the same tag is accepted again. +func TestDeleteTagHandler_ReleasesGuardAfterCompletion(t *testing.T) { + setupTestEnvironment() + withMockXdasSync(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + if isMetadataQuery(query) { + return []map[string]any{{"bucket_id": 1}}, nil + } + // Empty bucket: deletion completes immediately + return []map[string]any{}, nil + }) + + req := httptest.NewRequest("DELETE", "/taggingService/tags/guard-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "guard-tag"}) + rec := httptest.NewRecorder() + DeleteTagHandler(xwhttp.NewXResponseWriter(rec), req) + assert.Equal(t, http.StatusAccepted, rec.Code) + assert.Contains(t, rec.Body.String(), "queued for processing") + + // Wait for the background deletion to finish and release the guard. + // This must complete before the test ends so the mock DB is not restored + // under a still-running goroutine. + deletionKey := db.GetDefaultTenantId() + "|guard-tag" + deadline := time.Now().Add(5 * time.Second) + for { + if _, running := inFlightTagDeletions.Load(deletionKey); !running { + break + } + if time.Now().After(deadline) { + t.Fatal("background deletion did not release the in-flight guard") + } + time.Sleep(10 * time.Millisecond) + } +} + +// BUG-6: unknown tag on GET /tags/{tag} is a typed 404, not a string-matched +// error. +func TestGetTagByIdHandler_UnknownTagReturns404(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + return []map[string]any{}, nil + }) + + req := httptest.NewRequest("GET", "/taggingService/tags/missing-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "missing-tag"}) + w := httptest.NewRecorder() + GetTagByIdHandler(w, req) + + assert.Equal(t, http.StatusNotFound, w.Code) + assert.Contains(t, w.Body.String(), "missing-tag tag not found") +} + +// BUG-6: a database failure on GET /tags/{tag} is a 500, not a 404. +func TestGetTagByIdHandler_DbErrorReturns500(t *testing.T) { + setupTestEnvironment() + withMockDb(t, func(query string, params ...string) ([]map[string]any, error) { + return nil, errors.New("cassandra unavailable") + }) + + req := httptest.NewRequest("GET", "/taggingService/tags/some-tag", nil) + req = mux.SetURLVars(req, map[string]string{common.Tag: "some-tag"}) + w := httptest.NewRecorder() + GetTagByIdHandler(w, req) + + assert.Equal(t, http.StatusInternalServerError, w.Code) +} diff --git a/taggingapi/tag/tag_member_handler.go b/taggingapi/tag/tag_member_handler.go index 5b3d5af..db3d1cf 100644 --- a/taggingapi/tag/tag_member_handler.go +++ b/taggingapi/tag/tag_member_handler.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "strconv" + "sync" "github.com/rdkcentral/xconfadmin/adminapi/auth" "github.com/rdkcentral/xconfadmin/common" @@ -309,11 +310,6 @@ func GetTagByIdHandler(w http.ResponseWriter, r *http.Request) { tenantId := xhttp.GetTenantId(r) members, wasTruncated, err := GetTagById(tenantId, id) if err != nil { - // Check if tag not found - if err.Error() == "tag not found" { - xhttp.WriteXconfResponse(w, http.StatusNotFound, []byte(fmt.Sprintf(NotFoundErrorMsg, id))) - return - } xhttp.WriteXconfErrorResponse(w, err) return } @@ -342,6 +338,11 @@ func GetTagByIdHandler(w http.ResponseWriter, r *http.Request) { xhttp.WriteXconfResponse(w, statusCode, respBytes) } +// inFlightTagDeletions deduplicates concurrent background deletions of the +// same tag on this instance, keyed by "tenantId|tagId". Interim guard until +// tag deletion state is tracked cross-instance (tag registry, topic 3). +var inFlightTagDeletions sync.Map + // DeleteTagHandler deletes a tag and all its members from V2 storage asynchronously func DeleteTagHandler(w http.ResponseWriter, r *http.Request) { _, err := auth.CanWrite(r, auth.COMMON_ENTITY) @@ -374,8 +375,25 @@ func DeleteTagHandler(w http.ResponseWriter, r *http.Request) { return } + deletionKey := tenantId + "|" + id + if _, alreadyRunning := inFlightTagDeletions.LoadOrStore(deletionKey, true); alreadyRunning { + response := map[string]string{ + "status": "accepted", + "message": fmt.Sprintf("Tag '%s' deletion is already in progress", id), + "tag": id, + } + respBytes, err := json.Marshal(response) + if err != nil { + xhttp.WriteXconfErrorResponse(w, err) + return + } + xhttp.WriteXconfResponse(w, http.StatusAccepted, respBytes) + return + } + auditId := xw.AuditId() go func(tagId string) { + defer inFlightTagDeletions.Delete(deletionKey) if err := DeleteTag(tenantId, tagId); err != nil { log.WithFields(log.Fields{ "audit_id": auditId, diff --git a/taggingapi/tag/tag_member_service.go b/taggingapi/tag/tag_member_service.go index c72eb1d..06d2d42 100644 --- a/taggingapi/tag/tag_member_service.go +++ b/taggingapi/tag/tag_member_service.go @@ -158,11 +158,13 @@ func RemoveMembers(tenantId string, tagId string, members []string) error { allErrors = append(allErrors, fmt.Sprintf("bucket %d: %v", bucketId, err)) log.Errorf("Failed to remove %d members from bucket %d for tag %s: %v", len(bucketMembers), bucketId, tagId, err) - } else { - successCount += len(bucketMembers) - log.Debugf("Successfully removed %d members from bucket %d for tag %s", - len(bucketMembers), bucketId, tagId) + // The delete failed, so the bucket cannot have become empty — + // skip the metadata cleanup check + continue } + successCount += len(bucketMembers) + log.Debugf("Successfully removed %d members from bucket %d for tag %s", + len(bucketMembers), bucketId, tagId) // Clean up bucket metadata if bucket is now empty membersCount, err := getMembersCountOfBucket(tenantId, tagId, bucketId) if err != nil { @@ -417,20 +419,6 @@ func parseBucketedCursor(cursor string) (BucketedCursor, error) { return state, nil } -func min(a, b int) int { - if a < b { - return a - } - return b -} - -func max(a, b int) int { - if a > b { - return a - } - return b -} - // getReadWorkerCount returns the worker count for concurrent read operations func getReadWorkerCount() int { config := GetTagApiConfig() @@ -776,7 +764,7 @@ func GetTagById(tenantId string, tagId string) ([]string, bool, error) { } if len(populatedBuckets) == 0 { - return nil, false, fmt.Errorf("tag not found") + return nil, false, xwcommon.NewRemoteErrorAS(http.StatusNotFound, fmt.Sprintf(NotFoundErrorMsg, tagId)) } log.Infof("Fetching tag '%s' with %d populated buckets", tagId, len(populatedBuckets)) @@ -866,8 +854,11 @@ func deleteBucketMembers(tenantId string, tagId string, bucketId int) (int, erro } if len(removedFromXdas) < len(chunk) { - log.Warnf("partial XDAS deletion: %d/%d members removed", len(removedFromXdas), len(chunk)) - return totalDeleted, nil + // Partial XDAS failure must fail the bucket: returning success here + // would let DeleteTag report a completed deletion while leftover + // members and bucket metadata remain in both stores. + return totalDeleted, fmt.Errorf("partial XDAS deletion in bucket %d: %d/%d members removed", + bucketId, len(removedFromXdas), len(chunk)) } if len(chunk) < MaxBatchSizeV2 { diff --git a/taggingapi/tag/tag_member_service_bugs_test.go b/taggingapi/tag/tag_member_service_bugs_test.go index 6a966a4..21d4a74 100644 --- a/taggingapi/tag/tag_member_service_bugs_test.go +++ b/taggingapi/tag/tag_member_service_bugs_test.go @@ -20,23 +20,61 @@ import ( // mockDbClient injects Cassandra query behavior for service-level tests. // The embedded interface panics on any method the test does not stub, -// which flags unexpected database usage. +// which flags unexpected database usage. Batch and modify operations succeed +// by default; set the optional funcs to override. type mockDbClient struct { db.DatabaseClient - queryFunc func(query string, params ...string) ([]map[string]any, error) + queryFunc func(query string, params ...string) ([]map[string]any, error) + modifyFunc func(query string, params ...string) error + execBatchFn func(batch *mockBatch) error +} + +type mockBatch struct { + statements []string +} + +func (b *mockBatch) Query(stmt string, args ...any) { + b.statements = append(b.statements, stmt) +} + +func (b *mockBatch) Size() int { + return len(b.statements) } func (m *mockDbClient) QueryXconfDataRows(query string, params ...string) ([]map[string]any, error) { return m.queryFunc(query, params...) } -func withMockDb(t *testing.T, queryFunc func(query string, params ...string) ([]map[string]any, error)) { +func (m *mockDbClient) ModifyXconfData(query string, params ...string) error { + if m.modifyFunc != nil { + return m.modifyFunc(query, params...) + } + return nil +} + +func (m *mockDbClient) NewBatch(batchType int) db.BatchOperation { + return &mockBatch{} +} + +func (m *mockDbClient) ExecuteBatch(batch db.BatchOperation) error { + if m.execBatchFn != nil { + return m.execBatchFn(batch.(*mockBatch)) + } + return nil +} + +func withMockDbClient(t *testing.T, client *mockDbClient) { t.Helper() old := db.GetDatabaseClient() - db.SetDatabaseClient(&mockDbClient{queryFunc: queryFunc}) + db.SetDatabaseClient(client) t.Cleanup(func() { db.SetDatabaseClient(old) }) } +func withMockDb(t *testing.T, queryFunc func(query string, params ...string) ([]map[string]any, error)) { + t.Helper() + withMockDbClient(t, &mockDbClient{queryFunc: queryFunc}) +} + func isMetadataQuery(query string) bool { return strings.Contains(query, "tag_member_metadata") } From 089218d077e3d1a86dd6611beb517f89be93cd51 Mon Sep 17 00:00:00 2001 From: Maksym Dolina Date: Sun, 5 Jul 2026 20:39:56 +0100 Subject: [PATCH 4/4] Refactor inFlightTagDeletions variable placement for clarity in tag deletion handling --- taggingapi/tag/tag_member_handler.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/taggingapi/tag/tag_member_handler.go b/taggingapi/tag/tag_member_handler.go index db3d1cf..88bb406 100644 --- a/taggingapi/tag/tag_member_handler.go +++ b/taggingapi/tag/tag_member_handler.go @@ -23,6 +23,11 @@ const ( MaxPageSize = 5000 ) +// inFlightTagDeletions deduplicates concurrent background deletions of the +// same tag on this instance, keyed by "tenantId|tagId". Interim guard until +// tag deletion state is tracked cross-instance (tag registry, topic 3). +var inFlightTagDeletions sync.Map + func parsePaginationParams(r *http.Request) (*PaginationParams, error) { query := r.URL.Query() @@ -338,11 +343,6 @@ func GetTagByIdHandler(w http.ResponseWriter, r *http.Request) { xhttp.WriteXconfResponse(w, statusCode, respBytes) } -// inFlightTagDeletions deduplicates concurrent background deletions of the -// same tag on this instance, keyed by "tenantId|tagId". Interim guard until -// tag deletion state is tracked cross-instance (tag registry, topic 3). -var inFlightTagDeletions sync.Map - // DeleteTagHandler deletes a tag and all its members from V2 storage asynchronously func DeleteTagHandler(w http.ResponseWriter, r *http.Request) { _, err := auth.CanWrite(r, auth.COMMON_ENTITY)