Please address the comments from this code review:

## Overall Comments
- The VM index ACK resource ID separator is defined twice (`vmIndexACKResourceIDSeparator` in `central/sensor/service/common` and `vmACKResourceIDSeparator` in `compliance`); consider centralizing this into a shared constant to avoid accidental divergence between Central and Compliance.
- In `Relay.Run`, `r.umh.RetryCommand()` is called inline in the `select` on every loop iteration; it would be safer to obtain and store the retry channel once (and optionally guard against a nil `umh`) so that UMH implementations that might change or lazily initialize the channel cannot introduce subtle bugs.

## Individual Comments

### Comment 1
<location path="compliance/virtualmachines/relay/sender/index_report_sender.go" line_range="64-70" />
<code_context>
	result := "success"
</code_context>
<issue_to_address>
**suggestion (performance):** Using `prometheus.Labels` literals in hot paths causes per-call map allocations; prefer `WithLabelValues`.

Both metric calls create a new `map[string]string` on every send, which will show up in profiles under load. Use `WithLabelValues(result)` instead of `With(prometheus.Labels{...})` to avoid these allocations and follow the idiomatic pattern for fixed label sets.
</issue_to_address>

### Comment 2
<location path="compliance/virtualmachines/relay/relay.go" line_range="54" />
<code_context>
+	maxReportsPerMinute float64
+	staleAckThreshold   time.Duration
+
+	// cacheEvictTicker owns the metadata eviction ticker in production so it can be
+	// stopped when the relay shuts down, avoiding ticker goroutine leaks.
+	// Tests may leave this nil and inject cacheEvictTickCh directly.
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the relay’s caching and rate-limiting design by unifying ticker ownership, locking style, and cache invariants behind small helpers and a consistent time source.

You can reduce the added complexity without losing any functionality by tightening a few abstractions and making access patterns uniform.

### 1. Simplify ticker vs. channel ownership

You don’t need both `cacheEvictTicker` and `cacheEvictTickCh` on the struct. A simple injectable ticker factory keeps testability and removes the extra field and ownership questions.

```go
type Relay struct {
    // ...
    cacheEvictTicker *time.Ticker
    cache            map[string]*cachedReportMetadata
    mu               sync.Mutex
}

type tickerFactory func(d time.Duration) *time.Ticker

func defaultTickerFactory(d time.Duration) *time.Ticker {
    return time.NewTicker(d)
}

func New(
    reportStream IndexReportStream,
    reportSender sender.IndexReportSender,
    umh UnconfirmedMessageHandler,
    maxReportsPerMinute float64,
    staleAckThreshold time.Duration,
    newTicker tickerFactory, // nil -> default
) *Relay {
    if newTicker == nil {
        newTicker = defaultTickerFactory
    }

    r := &Relay{
        reportStream:        reportStream,
        reportSender:        reportSender,
        umh:                 umh,
        maxReportsPerMinute: maxReportsPerMinute,
        staleAckThreshold:   staleAckThreshold,
        cache:               make(map[string]*cachedReportMetadata),
    }

    if staleAckThreshold > 0 {
        interval := staleAckThreshold / 2
        if interval <= 0 {
            interval = staleAckThreshold
        }
        r.cacheEvictTicker = newTicker(interval)
    }

    umh.OnACK(r.markAcked)
    return r
}

func (r *Relay) Run(ctx context.Context) error {
    log.Info("Starting virtual machine relay")
    if r.cacheEvictTicker != nil {
        defer r.cacheEvictTicker.Stop()
    }

    reportChan, err := r.reportStream.Start(ctx)
    if err != nil {
        return errors.Wrap(err, "starting report stream")
    }

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case vmReport := <-reportChan:
            // ...
        case resourceID, ok := <-r.umh.RetryCommand():
            // ...
        case tick := <-r.cacheEvictTicker.C:
            r.evictStaleEntries(tick, r.staleAckThreshold)
        }
    }
}
```

Tests can inject a fake ticker implementation that exposes its own `C` channel without expanding the Relay’s public surface.

### 2. Use a single locking idiom

Mixing `concurrency.WithLock` and manual locking makes the concurrency story harder to follow. `markAcked` can match the rest of the file:

```go
func (r *Relay) markAcked(resourceID string) {
    metrics.AcksReceived.Inc()
    now := time.Now()

    r.mu.Lock()
    defer r.mu.Unlock()

    if cached, ok := r.cache[resourceID]; ok {
        cached.lastAckedAt = now
        return
    }
    r.cache[resourceID] = &cachedReportMetadata{
        updatedAt:   now,
        lastAckedAt: now,
    }
}
```

This also aligns with the rest of the cache helpers and makes invariants easier to reason about.

### 3. Centralize cache invariants behind helper methods

Right now the invariants for `updatedAt`, `lastAckedAt`, and `limiter` are spread across multiple methods, and `tryConsume` assumes the entry exists. You can make this safer and clearer by giving `cachedReportMetadata` small methods and centralizing “get or create” logic.

```go
type cachedReportMetadata struct {
    updatedAt   time.Time
    lastAckedAt time.Time
    limiter     *rate.Limiter
}

func (m *cachedReportMetadata) touch(now time.Time) {
    m.updatedAt = now
}

func (m *cachedReportMetadata) recordACK(now time.Time) {
    m.lastAckedAt = now
    if m.updatedAt.IsZero() {
        m.updatedAt = now
    }
}

func (m *cachedReportMetadata) ackStale(now time.Time, threshold time.Duration) bool {
    if m.lastAckedAt.IsZero() {
        return true
    }
    return now.Sub(m.lastAckedAt) > threshold
}

func (m *cachedReportMetadata) allow(now time.Time, maxPerMinute float64) bool {
    if maxPerMinute <= 0 {
        return true
    }
    if m.limiter == nil {
        ratePerSecond := maxPerMinute / 60.0
        m.limiter = rate.NewLimiter(rate.Limit(ratePerSecond), 1)
    }
    return m.limiter.Allow()
}

// getOrCreateMetadata is the single entry point for cache access.
func (r *Relay) getOrCreateMetadata(vsockID string, now time.Time) *cachedReportMetadata {
    r.mu.Lock()
    defer r.mu.Unlock()

    if m, ok := r.cache[vsockID]; ok {
        return m
    }
    m := &cachedReportMetadata{updatedAt: now}
    r.cache[vsockID] = m
    return m
}
```

Then `handleIncomingReport`, `markAcked`, and the staleness check all go through these helpers:

```go
func (r *Relay) markAcked(resourceID string) {
    metrics.AcksReceived.Inc()
    now := time.Now()

    r.mu.Lock()
    defer r.mu.Unlock()

    m, ok := r.cache[resourceID]
    if !ok {
        m = &cachedReportMetadata{}
        r.cache[resourceID] = m
    }
    m.recordACK(now)
}

func (r *Relay) handleIncomingReport(ctx context.Context, vmReport *v1.VMReport) {
    vsockID := vmReport.GetIndexReport().GetVsockCid()
    now := time.Now()

    m := r.getOrCreateMetadata(vsockID, now)
    r.mu.Lock()
    m.touch(now)
    r.mu.Unlock()

    if !m.allow(now, r.maxReportsPerMinute) {
        if r.staleAckThreshold > 0 && m.ackStale(now, r.staleAckThreshold) {
            // stale ACK path
        } else {
            // normal rate-limit path
        }
        return
    }

    // send and UMH calls as before
}
```

This makes the “caller must ensure the cache entry exists” comment unnecessary and prevents accidental nil dereferences if future callers skip `cacheReport`.

### 4. Use a consistent time source for related decisions

`handleIncomingReport` can pass `now` through to staleness checks, keeping eviction and ACK staleness based on the same time source:

```go
func (r *Relay) isACKStale(vsockID string, now time.Time) bool {
    r.mu.Lock()
    defer r.mu.Unlock()

    cached, ok := r.cache[vsockID]
    if !ok || cached.lastAckedAt.IsZero() {
        return true
    }
    return now.Sub(cached.lastAckedAt) > r.staleAckThreshold
}

func (r *Relay) handleIncomingReport(ctx context.Context, vmReport *v1.VMReport) {
    vsockID := vmReport.GetIndexReport().GetVsockCid()
    now := time.Now()

    r.cacheReport(vsockID, now)

    if !r.tryConsume(vsockID) {
        if r.staleAckThreshold > 0 && r.isACKStale(vsockID, now) {
            // ...
        } else {
            // ...
        }
        return
    }

    // ...
}
```

If you adopt the `cachedReportMetadata` helper methods above, you can avoid the extra map lookups entirely and keep “time” usage uniformly injected.

---

All of these changes keep the new behavior (rate limiting, ACK tracking, eviction, UMH wiring) but reduce cross-method invariants, clarify ownership of the ticker, and make the concurrency model easier to reason about.
</issue_to_address>

Please address the comments from this code review:

## Overall Comments
- The VM ACK resource ID separator is defined twice (`vmACKResourceIDSeparator` in `compliance` and `vmIndexACKResourceIDSeparator` in `central/sensor/service/common`); consider reusing a single shared constant to avoid accidental divergence in future changes.
- The `UnconfirmedMessageHandler.RetryCommand()` contract is relied on to return a stable channel in `Relay.Run`; it would be safer either to document this requirement on the interface or to retrieve the channel once in `New` and store it on the Relay, to avoid unexpected behavior if an implementation changes to allocate channels lazily.

## Individual Comments

### Comment 1
<location path="compliance/virtualmachines/relay/sender/index_report_sender.go" line_range="64-69" />
<code_context>
+	result := "success"
</code_context>
<issue_to_address>
**suggestion (performance):** Use WithLabelValues instead of With(prometheus.Labels{...}) for high-frequency metrics to reduce allocations

This send path is relatively hot, and calling `.With(prometheus.Labels{"result": result})` for both `VMIndexReportSendAttempts` and `VMIndexReportSendDurationSeconds` allocates a new labels map each time. Since the label key is fixed, using `.WithLabelValues(result)` for both metrics avoids those per-call allocations and aligns with Prometheus client best practices.

Suggested implementation:

```golang
	metrics.VMIndexReportSendAttempts.WithLabelValues(result).Inc()

```

```golang
	timer := prometheus.NewTimer(metrics.VMIndexReportSendDurationSeconds.WithLabelValues(result))

```

These replacements assume that `VMIndexReportSendAttempts` and `VMIndexReportSendDurationSeconds` are declared with a single label (e.g. `prometheus.NewCounterVec(..., []string{"result"})`). If they have multiple labels, you will need to pass all label values to `WithLabelValues(...)` in the correct order, not just `result`.
</issue_to_address>