[9.5](backport #51801) filestream: fix duplicate harvesters on growing fingerprint migration

[mergify]

Proposed commit message

With file_identity.fingerprint.growing enabled, the registry key
migration that follows a file's fingerprint growth raced with
asynchronously starting harvesters. Start reserves the source id
synchronously, but the harvester goroutine only registers and locks its
resource when it gets to run. A migration landing in that window ignored
the reserved-but-unregistered harvester, so the follow-up Start of the
new key spawned a second harvester for the same file, while the pending
one registered under the stale key, locked a fresh offset-0 resource,
and re-read the whole file. Every line was ingested once per extra
harvester (2x-3x observed) and the migrated-away keys were resurrected
in the registry with cursor data.

The reader bookkeeping now stores a handle from reservation on and the
whole identity chain moves atomically:

- reserve() returns a *reader handle; the goroutine upgrades it via
  register(), which follows any re-keying that happened in the meantime
  and fails if the reservation was removed, so a Stop that arrives
  before the harvester starts now sticks.
- HarvesterGroup.Migrate re-keys the registry entry (UpdateKey) and any
  registration in one critical section, and derives the new key from the
  Source so it always matches what a follow-up Start reserves.
- Harvesters read their (key, resource) pair atomically; because
  UpdateKey re-keys resources in place, the pair stays live across
  migrations and a stale key can no longer mint a fresh resource.
- UpdateKey rolls back its in-memory swap when persisting fails, so a
  failed migration leaves a consistent old-keyed world; the prospector
  keeps the file flowing under its old identity and retries on a later
  scan, pruning index entries whose registry key is gone (ErrKeyGone).
- Stop is synchronous so a queued Stop can never cancel a newer Start's
  reservation, and stopping is no longer delayed behind the
  harvester_limit semaphore.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• [ ] I have made corresponding changes to the documentation
• [ ] I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works. Where relevant, I have used the stresstest.sh script to run them under stress conditions and race detector to verify their stability.
• [ ] I have added an entry in ./changelog/fragments using the changelog tool. (skip-changelog: the growing fingerprint feature is unreleased)

Disruptive User Impact

None.

How to test this PR locally

Unit and integration tests:

go test -race ./filebeat/input/filestream/...
go test -tags integration ./filebeat/input/filestream/...

Reproduce with harvester_limit which keeps starting harvesters queued on the task group semaphore.

filebeat.inputs:
  - type: filestream
    id: repro-growing
    paths: ["/tmp/gfp/logs/*.log"]
    prospector.scanner.check_interval: 100ms
    prospector.scanner.fingerprint.enabled: true
    prospector.scanner.fingerprint.offset: 0
    prospector.scanner.fingerprint.length: 1024
    file_identity.fingerprint.growing: true
    harvester_limit: 10
    close.on_state_change.inactive: 500ms
output.file:
  path: /tmp/gfp/out
  filename: events

Create ~30 files with ~600 bytes (below the 1024-byte fingerprint threshold), start filebeat, then append lines in two waves so each file first grows below the threshold and then crosses it. Group the output docs by (log.file.path, log.offset): before this fix files that were queued behind the limit get every line 2-3 times (and filebeat.harvester.started exceeds the file count); after it every pair appears exactly once.

---

This is an automatic backport of pull request #51801 done by [Mergify](https://mergify.com).

[github-actions]

🤖 GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)
• /test : Run the Buildkite pipeline.

[infra-vault-gh-plugin-prod]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)