Ensure deterministic outcomes when multiple create listener APIs are racing

[peterbroadhurst]

When multiple APIs are creating/update a listener concurrently under an event stream, there are cases where we could return the UUID of one running listener, but actually internally in the runtime be running a different one.

This PR works through a number of sub-items to protect in these situations.

Duplicate listener names were not checked until the database write

createOrUpdateListener called AddOrUpdateListener first, which inserted into the stream's in-memory listener map and called connector.EventListenerAdd starting the listener in the connect.

Only afterwards did WriteListener attempt the insert in Postgres, where there is alisteners_name unique index.

Two parallel POST /eventstreams/{id}/listeners with the same name were therefore both registered and started in the connector before either was checked.

Outcomes:

• Postgres — the loser got an opaque FF00177 500, both listeners were live in the connector concurrently (duplicate event delivery), and the cleanup was a best-effort RemoveListener whose error was only logged.
• LevelDB — no name index at all, so both succeeded and the duplicate was permanent.

Note an empty name defaults to the connector-resolved signature, so two parallel creates with identical filters and no name collide deterministically.

Runtime and connector state were updated before DB write

Any failure after AddOrUpdateListener had to be unwound by removing the listener again, which is correct for a create but wrong for an update.
lockedListenerUpdate had already overwritten l.spec in place, so the rollback deleted a pre-existing listener rather than restoring it.

Fix

Split AddOrUpdateListener into resolve / prepare / apply, so the manager can persist between validating and applying:

• VerifyListenerOptions resolves the full spec (including the defaulted name) without touching runtime state or the connector.
• PrepareListenerUpdate validates the spec against the stream and returns a PreparedListenerUpdate; Apply() makes it live.
• AddOrUpdateListener keeps its signature as a thin wrapper over the three steps, so connectors and existing callers are unaffected.

New error FF21098 (Duplicate listener name '%s' used by listener '%s') returns 409 where callers previously saw a 500 or silent duplication.

Uniqueness is still global across streams, matching the Postgres index. No change to that behavior in this PR.

Names are released only after the persistence delete succeeds, and every release is guarded on the ID still holding the name, so a losing racer can never evict the winner's reservation.

Startup

restoreStreams populates the name index from persisted listeners. Pre-existing LevelDB duplicates warn and first-wins rather than failing startup (a hard failure would turn an upgrade into an unrecoverable boot loop).

[Chengxuan]

Given the break up of AddOrUpdateListener function. The Prepare and Apply steps are no longer under the same lock.

This creates a race window for any other function that mutates the map while persistence.WriteListener is running.

Therefore a concurrent delete listener request from a user can complete in that gap, but the listener will get recreated if the u (PreparedListenerUpdate) contains it.

[peterbroadhurst]

Yes, that is correct.

This PR I assert makes things significantly better not worse, as before the window was we didn't lock between writing potential future config to the map, and then attempting the DB update. The fundamental issue.

Now, we have two phases. Deliberately separated as with the existing locks they cannot be in the same scope.

1. Prepare a safe change to the in-memory structure, without applying
2. Update the database - which is the point of control for duplicate
3. If successful, apply the safe change to the in-memory structure

The dual insert that was the root issue, is fully protected by this change. Hence the proposal for it.

You are correct that when I was writing this, I was concerned about the fact that a different competing action could mutate the database separately.

The key one I looked at was a delete. The other would be an edit that makes a different change - such as a dual reset to different from blocks.

The feedback is fair, and I'll look if there's a low-risk enhancement I can make to the code to cover more cases.

[peterbroadhurst]

Thanks again for the feedback, I went back and established it is safe from a lock hierarchy perspective to introduce a course grain lock on admin functions (without affecting the runtime scoping covered in the primary change).

Mind taking another look?

6fa42c0

[Chengxuan]

LGTM.

[Chengxuan]

Thanks for adding the extra admin mutex. Looks great!