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..88bb406 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" @@ -22,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() @@ -309,11 +315,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 } @@ -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") }