[filestream-registry] clean_removed can delete active filestream state before ACK

## Finding

`clean_removed` can remove a filestream registry entry while its harvester is still running and while cursor updates are still in flight. This can silently lose the only durable resume point for bytes that were read from an unlinked file but not yet safely persisted, and it also causes the pending ACK operation to return before releasing its resource ownership.

## Evidence

Defaults make this reachable on normal Linux filestream configurations: `clean_removed` defaults to `true`, while `close.on_state_change.removed` defaults to `false`.

```go
// filebeat/input/filestream/config.go:165-166
CleanInactive: -1,
CleanRemoved: true,
```

```go
// filebeat/input/filestream/defaultconfig.go:22-25
// defaultCloserOnStateChangeRemoved returns the default configuration value for
// close.on_state_change.removed
func defaultCloserOnStateChangeRemoved() bool {
return false
}
```

On delete, the prospector removes state immediately. It only stops the harvester if `close.on_state_change.removed` is enabled, and even then `hg.Stop` is asynchronous and the removal happens immediately afterward.

```go
// filebeat/input/filestream/prospector.go:448-458
func (p *fileProspector) onRemove(log *logp.Logger, fe loginp.FSEvent, src loginp.Source, s loginp.StateMetadataUpdater, hg loginp.HarvesterGroup) {
if p.stateChangeCloser.Removed {
log.Debugf("Stopping harvester as file %s has been removed and close.on_state_change.removed is enabled.", src.Name())
hg.Stop(src)
}

if p.cleanRemoved {
log.Debugf("Remove state for file as file removed: %s", fe.OldPath)

err := s.Remove(src)
```

`Remove` does not acquire the per-resource harvester lock or check pending cursor operations; it sets TTL 0 on whatever resource is found.

```go
// filebeat/input/filestream/internal/input-logfile/store.go:602-609
// Removes marks an entry for removal by setting its TTL to zero.
func (s *store) remove(key string) error {
resource := s.ephemeralStore.Find(key, false)
if resource == nil {
return fmt.Errorf("resource '%s' not found", key)
}
s.UpdateTTL(resource, 0)
resource.Release()
```

TTL 0 immediately persists a deleted state, increments the resource version, and invalidates it so later ACK cursor writes cannot restore it.

```go
// filebeat/input/filestream/internal/input-logfile/store.go:618-641
// If the TTL of the resource is set to 0, once it is persisted, it is going to be removed from the
// store in the next cleaner run. The resource also gets invalidated to make sure new updates are not
// saved to the registry.
func (s *store) UpdateTTL(resource *resource, ttl time.Duration) {
...
if resource.unsafeIsDeleted() {
// version must be incremented to make sure existing resource
// instances do not overwrite the removal of the entry
resource.version++
// invalidate it after it has been persisted to make sure it cannot
// be overwritten in the persistent store
resource.invalid = true
```

When the output ACK later arrives, `Execute` returns before applying the cursor and before `op.done(n)` is deferred, because the resource version changed or the resource is deleted.

```go
// filebeat/input/filestream/internal/input-logfile/publish.go:119-130
func (op *updateOp) Execute(store *store, n uint) {
resource := op.resource

resource.stateMutex.Lock()
defer resource.stateMutex.Unlock()

if resource.lockedVersion != op.resource.version || resource.unsafeIsDeleted() {
return
}

defer op.done(n)
resource.activeCursorOperations -= n
```

The repository already contains a related integration-test comment documenting that a perceived remove event deletes state and corrupts subsequent tracking:

```go
// filebeat/input/filestream/input_integration_test.go:116-124
// On a flaky execution, the file is actually perceived as removed
// and then a new file is created, both with the same inode. This
// happens on a system that does not reuse inodes as soon they're
// freed. Because the file is detected as removed, it's state is also
// removed. Then when more data is added, only the offset of the new
// data is tracked by the registry, causing the test to fail.
//
// A workaround for this is to not remove the state when the file is
// removed, hence `clean_removed: false` is set here.
```

## User-triggerable reproduction

1. Configure a Linux filestream input with defaults (`clean_removed: true`, `close.on_state_change.removed: false`) and a slow or blocked output so events remain unACKed.
2. Write a large log file matched by the input.
3. While Filebeat is harvesting the file and the output is blocked, delete/unlink the file (`rm app.log`) or perform a rotation pattern that the watcher reports as delete/create.
4. The delete event calls `onRemove` and `Remove`, which deletes/invalidates the registry state even though the harvester can still read the open inode and cursor ACKs are pending.
5. Kill Filebeat before the blocked output safely ACKs all published events. The unlinked file cannot be reopened, and the registry no longer contains the last safe cursor, causing silent loss of bytes read after the last persisted cursor.

A focused unit test can be added under `filebeat/input/filestream/internal/input-logfile/store_test.go` or `publish_test.go`:

- create a stored resource with offset 10;
- simulate an active harvester lock with `lock(...)`;
- create a pending update op for offset 20 via `createUpdateOp`;
- call `store.remove(key)`;
- execute the ACK op;
- assert the persistent cursor is still not advanced to 20 and the resource remains pending because `op.done` was skipped.

Scoped test command:

```bash
go test ./filebeat/input/filestream/internal/input-logfile -run 'Test.*Remove.*PendingACK|Test.*CleanRemoved.*Active'
```

## Suggested fix

Do not let `clean_removed` invalidate/remove a resource that is actively harvested or has pending cursor operations. Make `Remove` follow the same safety model as cleanup:

- acquire/try the resource lock before marking TTL 0, or otherwise detect active ownership/pending cursor operations;
- if active, defer deletion until the harvester exits and all ACK updates have been written;
- ensure `updateOp.Execute` always releases ownership with `op.done(n)` even when dropping a stale/deleted update, to avoid leaking pending ownership;
- add regression tests for delete while harvester active, delete with pending ACK, and delete with `close.on_state_change.removed: true` racing asynchronous stop.

---
[What is this?](https://ela.st/github-ai-tools) | [From workflow: Sweeper: Filestream Registry and State Machine](https://github.com/elastic/beats/actions/runs/28436946784)

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.
> - [x] expires on Jul 7, 2026, 10:23 AM UTC

Start with onRemove in filebeat/input/filestream/prospector.go, then read Remove and UpdateTTL in filebeat/input/filestream/internal/input-logfile/store.go and updateOp.Execute in publish.go. Run the scoped input-logfile tests and add coverage for removal during an active harvester or pending ACK. Done means removal cannot invalidate active or pending state, ACK ownership is released, and the regression cases pass.

Written by the indexing model from the issue text.