then:
        executedAndNotSkipped ':sync'
    }

    @NotYetImplemented
    def 'sync is not up to date when files are added to the destination dir'() {
        given:
        defaultSourceFileTree()
        file('dest').create {}

        buildScript '''
            task sync(type: Sync) {
                from 'source'
                into 'dest'
            }

import "istio.io/istio/pkg/kube"

type Syncer interface {
	WaitUntilSynced(stop <-chan struct{}) bool
	HasSynced() bool
}

var (
	_ Syncer = channelSyncer{}
	_ Syncer = pollSyncer{}
)

type channelSyncer struct {
	name   string
	synced <-chan struct{}
}

func (c channelSyncer) WaitUntilSynced(stop <-chan struct{}) bool {
	return waitForCacheSync(c.name, stop, c.synced)
}

	return res
}

func (j *join[T]) Register(f func(o Event[T])) Syncer {
	return registerHandlerAsBatched[T](j, f)
}

func (j *join[T]) RegisterBatch(f func(o []Event[T], initialSync bool), runExistingState bool) Syncer {
	sync := multiSyncer{}
	for _, c := range j.collections {
		sync.syncers = append(sync.syncers, c.RegisterBatch(f, runExistingState))
	}
	return sync
}

// nolint: unused // (not true, its to implement an interface)

	return maps.Values(s.vals)
}

func (s *staticList[T]) Register(f func(o Event[T])) Syncer {
	return registerHandlerAsBatched(s, f)
}

func (s *staticList[T]) Synced() Syncer {
	return alwaysSynced{}
}

func (s *staticList[T]) RegisterBatch(f func(o []Event[T], initialSync bool), runExistingState bool) Syncer {
	if runExistingState {
		f(slices.Map(s.List(), func(e T) Event[T] {
			return Event[T]{

type multiIndex[I, O any] struct {
	outputs  map[Key[O]]O
	inputs   map[Key[I]]I
	mappings map[Key[I]]sets.Set[Key[O]]
}

func (h *manyCollection[I, O]) Synced() Syncer {
	return channelSyncer{
		name:   h.collectionName,
		synced: h.synced,
	}
}

// nolint: unused // (not true, its to implement an interface)
func (h *manyCollection[I, O]) dump() {
	h.recomputeMu.Lock()
	defer h.recomputeMu.Unlock()

var _ internalCollection[controllers.Object] = &informer[controllers.Object]{}

func (i *informer[I]) _internalHandler() {}

func (i *informer[I]) Synced() Syncer {
	return channelSyncer{
		name:   i.collectionName,
		synced: i.synced,
	}
}

// nolint: unused // (not true, its to implement an interface)
func (i *informer[I]) dump() {
	i.log.Errorf(">>> BEGIN DUMP")
	for _, obj := range i.inf.List(metav1.NamespaceAll, klabels.Everything()) {

	pvcListerSynced cache.InformerSynced
	// revListerSynced returns true if the rev shared informer has synced at least once
	revListerSynced cache.InformerSynced
	// StatefulSets that need to be synced.
	queue workqueue.TypedRateLimitingInterface[string]
	// eventBroadcaster is the core of event processing pipeline.
	eventBroadcaster record.EventBroadcaster
}

	// using krt), registering directly is expected.
	Register(f func(o Event[T])) Syncer

	// Synced returns a Syncer which can be used to determine if the collection has synced. Once its synced, all dependencies have
	// been processed, and all handlers have been called with the results.
	Synced() Syncer

	// RegisterBatch registers a handler that accepts multiple events at once. This can be useful as an optimization.

		},
	)

	// SyncProxyRulesLastTimestamp is the timestamp proxy rules were last
	// successfully synced.
	SyncProxyRulesLastTimestamp = metrics.NewGauge(
		&metrics.GaugeOpts{
			Subsystem:      kubeProxySubsystem,
			Name:           "sync_proxy_rules_last_timestamp_seconds",
			Help:           "The last time proxy rules were successfully synced",
			StabilityLevel: metrics.ALPHA,
		},
	)

	v := d.val.Load()
	if v == nil {
		return nil
	}
	return []T{*v}
}

func (d *static[T]) Register(f func(o Event[T])) Syncer {
	return registerHandlerAsBatched[T](d, f)
}

func (d *static[T]) RegisterBatch(f func(o []Event[T], initialSync bool), runExistingState bool) Syncer {
	d.eventHandlers.Insert(f)
	if runExistingState {
		v := d.val.Load()
		if v != nil {
			f([]Event[T]{{
				New:   v,