fix(shard-distributor): make DeleteShardStats non-transactional and fix cleanup condition

DeleteShardStats doesn't need transactional guards since we are just working with telemetry.

Signed-off-by: Andreas Holt <[email protected]>

Author: AndreasHolt

service/sharddistributor/leader/process/processor.go

@@ -237,11 +237,11 @@ func (p *namespaceProcessor) runShardStatsCleanupLoop(ctx context.Context) {
 				continue
 			}
 			staleShardStats := p.identifyStaleShardStats(namespaceState)
-			if len(staleShardStats) > 0 {
+			if len(staleShardStats) == 0 {
 				// No stale shard stats to delete
 				continue
 			}
-			if err := p.shardStore.DeleteShardStats(ctx, p.namespaceCfg.Name, staleShardStats, p.election.Guard()); err != nil {
+			if err := p.shardStore.DeleteShardStats(ctx, p.namespaceCfg.Name, staleShardStats); err != nil {
 				p.logger.Error("Failed to delete stale shard stats", tag.Error(err))
 			}
 		}

service/sharddistributor/store/etcd/executorstore/etcdstore.go

@@ -625,37 +625,19 @@ func (s *executorStoreImpl) DeleteExecutors(ctx context.Context, namespace strin
 	return nil
 }
 
-func (s *executorStoreImpl) DeleteShardStats(ctx context.Context, namespace string, shardIDs []string, guard store.GuardFunc) error {
+func (s *executorStoreImpl) DeleteShardStats(ctx context.Context, namespace string, shardIDs []string) error {
 	if len(shardIDs) == 0 {
 		return nil
 	}
-	var ops []clientv3.Op
+
 	for _, shardID := range shardIDs {
 		shardStatsKey, err := etcdkeys.BuildShardKey(s.prefix, namespace, shardID, etcdkeys.ShardStatisticsKey)
 		if err != nil {
 			return fmt.Errorf("build shard statistics key: %w", err)
 		}
-		ops = append(ops, clientv3.OpDelete(shardStatsKey))
-	}
-
-	nativeTxn := s.client.Txn(ctx)
-	guardedTxn, err := guard(nativeTxn)
-
-	if err != nil {
-		return fmt.Errorf("apply transaction guard: %w", err)
-	}
-	etcdGuardedTxn, ok := guardedTxn.(clientv3.Txn)
-	if !ok {
-		return fmt.Errorf("guard function returned invalid transaction type")
-	}
-
-	etcdGuardedTxn = etcdGuardedTxn.Then(ops...)
-	resp, err := etcdGuardedTxn.Commit()
-	if err != nil {
-		return fmt.Errorf("commit shard statistics deletion: %w", err)
-	}
-	if !resp.Succeeded {
-		return fmt.Errorf("transaction failed, leadership may have changed")
+		if _, err := s.client.Delete(ctx, shardStatsKey); err != nil {
+			return fmt.Errorf("delete shard statistics: %w", err)
+		}
 	}
 	return nil
 }

service/sharddistributor/store/store.go

@@ -63,7 +63,7 @@ type Store interface {
 	AssignShards(ctx context.Context, namespace string, request AssignShardsRequest, guard GuardFunc) error
 	Subscribe(ctx context.Context, namespace string) (<-chan int64, error)
 	DeleteExecutors(ctx context.Context, namespace string, executorIDs []string, guard GuardFunc) error
-	DeleteShardStats(ctx context.Context, namespace string, shardIDs []string, guard GuardFunc) error
+	DeleteShardStats(ctx context.Context, namespace string, shardIDs []string) error
 
 	GetShardOwner(ctx context.Context, namespace, shardID string) (*ShardOwner, error)
 	SubscribeToAssignmentChanges(ctx context.Context, namespace string) (<-chan map[*ShardOwner][]string, func(), error)