fix: Cleanup deleted and disabled sinks

Author: jachym-tousek-keboola

internal/pkg/service/stream/definition/repository/sink/sink_watch.go

@@ -12,3 +12,7 @@ import (
 func (r *Repository) GetAllAndWatch(ctx context.Context, opts ...etcd.OpOption) *etcdop.RestartableWatchStreamT[definition.Sink] {
 	return r.schema.Active().GetAllAndWatch(ctx, r.client, opts...)
 }
+
+func (r *Repository) GetActivePlusDeletedAndWatch(ctx context.Context, opts ...etcd.OpOption) *etcdop.RestartableWatchStreamT[definition.Sink] {
+	return r.schema.GetAllAndWatch(ctx, r.client, opts...)
+}

internal/pkg/service/stream/storage/node/coordinator/metacleanup/metacleanup.go

@@ -101,7 +101,7 @@ func Start(d dependencies, cfg Config) error {
 	})
 
 	n.sinks = etcdop.SetupMirrorMap[definition.Sink, key.SinkKey, *sinkData](
-		n.definitionRepository.Sink().GetAllAndWatch(ctx),
+		n.definitionRepository.Sink().GetActivePlusDeletedAndWatch(ctx),
 		func(_ string, sink definition.Sink) key.SinkKey {
 			return sink.SinkKey
 		},
@@ -165,10 +165,6 @@ func (n *Node) cleanMetadata(ctx context.Context) (err error) {
 
 	// Process all sink keys
 	n.sinks.ForEach(func(sinkKey key.SinkKey, sink *sinkData) (stop bool) {
-		if !sink.Enabled {
-			return false
-		}
-
 		grp.Go(func() error {
 			// There can be several cleanup nodes, each node processes an own part.
 			owner, err := n.dist.IsOwner(sinkKey.ProjectID.String())