[filestream-registry] Filestream migration and ACK persistence can regress cursor and re-ingest data

## Findings

### 1. Identifier migration ignores parser suffix, so identity changes can skip state migration and re-read files from offset 0

**Priority:** P1 (likely in production when users change `file_identity` with parser stream suffix enabled)

**Location**
- `filebeat/input/filestream/prospector.go:54-72`
- `filebeat/input/filestream/prospector.go:275-297`
- `filebeat/input/filestream/prospector_creator.go:99-103`
- `filebeat/input/filestream/prospector_creator.go:130`
- `filebeat/input/filestream/identifier.go:189-192`

**Evidence**
- Migration key check uses `identifiersMap[oldIdentifierName]` from package init (`prospector.go:54-72`) and computes `previousIdentifierKey` (`prospector.go:288-293`).
- That map is built without parser suffix, while runtime identifiers are created with suffix from `config.Reader.Parsers.Suffix` (`prospector_creator.go:99-103,130`).
- Suffix is part of identity (`fs.fileID += "-" + s.suffix`) in `identifier.go:189-192`.
- Result: `previousIdentifierKey != registryKey` path (`prospector.go:296-297`) causes migration skip.

**What is wrong**
When `file_identity` is changed (e.g., native/path -> fingerprint) and parser suffix is set, migration compares unsuffixed keys against suffixed registry keys and skips valid entries.

**Why it matters**
Skipped migration leaves old cursor under old identity while new identity starts with no state, so files are harvested from the beginning and duplicate already-sent data.

**Reproduction scenario**
1. Run filestream with parser suffix enabled (for example stream suffix) and `file_identity: native`; ingest data.
2. Change config to `file_identity: fingerprint`; restart Filebeat.
3. Observe migration skip path (`previousIdentifierKey != registryKey`), no key move, and re-ingestion from beginning under new identity.

**Fix direction**
Build migration-time old identifiers with the same parser suffix as runtime identifiers (or canonicalize key comparison to a suffix-aware identity form).

---

### 2. `take_over.stream` empty (documented as "all streams") cannot match stream-hashed log states

**Priority:** P1 (common migration path for container/stdout+stderr users)

**Location**
- `filebeat/input/filestream/internal/input-logfile/manager.go:346-347`
- `filebeat/input/filestream/prospector.go:109-115`
- `filebeat/input/filestream/prospector.go:161-166`
- `filebeat/input/filestream/internal/input-logfile/store.go:618-621`
- `filebeat/input/filestream/internal/input-logfile/store.go:633-637`
- `filebeat/input/file/identifier.go:99-114`

**Evidence**
- Config comment says empty stream means all streams (`manager.go:346-347`).
- `previousID` injects `Meta["stream"]` only when stream is explicitly `stdout` or `stderr` (`prospector.go:109-115`).
- Matching requires exact ID equality (`prospector.go:161-166`).
- Log identifier IDs include hashed `Meta` when present (`file/identifier.go:99-114`).
- `newTakeOverState` for log states carries `Source`, `IdentifierName`, `FileStateOS`, `Offset` but not `Meta` (`store.go:618-621,633-637`).

**What is wrong**
For log states whose ID was stream-hashed, `take_over.stream: ""` computes an ID without stream metadata and fails equality check, so takeover skips valid states.

**Why it matters**
Users migrating to filestream with empty stream (expecting "all") can lose takeover continuity and re-ingest large existing container logs from offset 0.

**Reproduction scenario**
1. Have log/container states on disk for both stdout/stderr with stream metadata in IDs.
2. Configure filestream takeover with `take_over.enabled: true` and empty `take_over.stream`.
3. Restart; takeover callback returns empty for those entries due ID mismatch; filestream creates fresh state and re-reads content.

**Fix direction**
Either preserve/propagate stream metadata into `TakeOverState` and matching, or when stream is empty attempt matching both stream variants (stdout/stderr) for log-source takeover.

---

### 3. ACKed cursor updates are dropped permanently on registry write errors

**Priority:** P2 (high impact when disk/registry backend is transiently unavailable)

**Location**
- `filebeat/input/filestream/internal/input-logfile/publish.go:131-151`
- `filebeat/input/filestream/internal/input-logfile/update_writer.go:96-100`

**Evidence**
- ACK operation executes `store.persistentStore.Set(...)`; on error it only logs (`publish.go:147-151`).
- The op is finalized in the same execution (`defer op.done(n)`) and not requeued (`publish.go:131`).
- Writer loop just iterates received batch once (`update_writer.go:96-100`) and has no retry/backoff path for failed `Set`.

**What is wrong**
If registry write fails after ACK (e.g., ENOSPC/temporary FS error), the durable cursor advance is lost with no retry.

**Why it matters**
A restart after transient registry-write failure replays already-ACKed data from the last persisted offset, causing duplicates and potential downstream overload.

**Reproduction scenario**
1. Run filestream and generate events.
2. Induce temporary registry write failure (e.g., registry FS full/read-only) while events are ACKing.
3. Restore FS and restart Filebeat without additional writes on that file.
4. Observe resume from stale offset and duplicate replay.

**Fix direction**
Retain failed ACK updates for retry (or force input degradation/shutdown until persistence succeeds) so ACK cannot be considered complete until cursor durability is re-established.

## Scenarios investigated and found safe
- Cleaner does not remove active/in-flight resources: `checkCleanResource` requires `resource.Finished()` and `resource.stored` (`filebeat/input/filestream/internal/input-logfile/clean.go:115-131`).
- TTL=0 invalidation prevents stale ACK overwrite after deletion (`filebeat/input/filestream/internal/input-logfile/store.go:762-769`, `filebeat/input/filestream/internal/input-logfile/publish.go:127-129`).
- Growing-fingerprint migration failure intentionally keeps old identity to avoid split state (`filebeat/input/filestream/prospector.go:637-642`).

## Suggested Actions
- [ ] Make identity-change migration suffix-aware so old-key matching works when parser suffix is configured.
- [ ] Fix empty-stream takeover matching for stream-hashed log states so "all streams" behaves as documented.
- [ ] Add durable retry/guaranteed handling for ACKed cursor writes that fail `persistentStore.Set`.

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

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.
> - [x] expires on Sep 1, 2026, 9:25 AM UTC

Start with the migration paths in filebeat/input/filestream/prospector.go, prospector_creator.go, and identifier.go, then inspect takeover matching in internal/input-logfile/manager.go and store.go. Read ACK handling in internal/input-logfile/publish.go and update_writer.go, and reproduce the three scenarios described. Done means identity migration, empty-stream takeover, and failed ACK persistence all retain cursor continuity without re-ingestion.

Written by the indexing model from the issue text.