fix(streams): make StreamQueue.record! independent of Rails schema-cache index resolution

[mhenrixon]

Summary

• Pgbus::StreamQueue.record! (app/models/pgbus/stream_queue.rb) no longer uses upsert(unique_by: :queue_name) — it writes via raw INSERT INTO … (queue_name) VALUES (…) ON CONFLICT (queue_name) DO NOTHING on the model's connection. The unique index is owned by the gem's own migration, so there is nothing for Rails to resolve, and the pool schema cache's permanently-latched negative data_source_exists? probe can no longer poison registration for the process lifetime.
• Failure logging is class-aware: transient database errors (ActiveRecord::ActiveRecordError) log at DEBUG per attempt as before; any other StandardError — the bug-signal class the old path's ArgumentError belonged to — logs at WARN once per process, DEBUG thereafter.
• Integration harness now bootstraps pgbus_stream_queues (mirroring the generator template) so the registry is exercised for real, and a new spec/integration/stream_queue_registry_spec.rb pins the fix.

Closes #401

Acceptance criteria → proof

Criterion	Proof
record! succeeds with a deliberately poisoned pool schema cache	spec/integration/stream_queue_registry_spec.rb poisons @data_sources["pgbus_stream_queues"] = false exactly as the issue's repro; verified RED against the old implementation (fails) and GREEN with the fix
No schema-cache indexes/data_source_exists? query from record!	sql.active_record subscription asserts zero SCHEMA-named events during record! once the table_exists? memo is warm
Resolution-class failure logs at WARN at most once per process	Unit specs: ArgumentError → one WARN then DEBUG, across different queue names; StatementInvalid → DEBUG only, never WARN
backfill! / all_names / cache-consistency semantics unchanged	Existing specs untouched and green (warm/cold cache, backfill counting, return values)

Test plan

• bundle exec rspec spec/pgbus/ spec/generators/ — 3866 examples, 0 failures
• bundle exec rake rubocop — 553 files, no offenses
• spec/integration/stream_queue_registry_spec.rb — poisoned-cache regression fails on old code, passes on new
• Full integration suite against local PG

Deviations & judgment calls

• Issue point 3 (move the registry write off the first-broadcast hot path) not implemented — it's phrased as "consider", the raw-SQL fix already removes all schema-cache/InsertAll traffic from the flush thread, and the write is already once-per-stream-per-process via the @stream_indexes_created memo. Deferral would change ensure_stream_queue semantics and belongs in its own PR if still wanted.
• Issue point 4 (first-broadcast hang audit) is explicitly a separate follow-up per the issue — out of scope here.
• Transient-vs-bug error split implemented as ActiveRecord::ActiveRecordError → DEBUG (covers StatementInvalid and connection errors like ConnectionNotEstablished, which are equally transient), any other StandardError → WARN once per process then DEBUG. The issue said "non-StatementInvalid", but that literal reading would WARN on transient connection drops.
• Used connection.quote/quote_table_name interpolation instead of the issue's sanitize_sql([… ?]) — in Rails 8.1 sanitize_sql_array checks out a pool connection via with_connection, which can't be stubbed in the DB-free unit suite; connection-level quoting is equivalent and testable through the same connection seam the spec file already stubs.

---

Summary by cubic

Makes Pgbus::StreamQueue.record! write via raw INSERT…ON CONFLICT instead of upsert(unique_by: :queue_name), removing dependence on Rails’ pool schema cache and preventing process-lifetime failures after a false negative probe.

• No schema-cache index resolution or SCHEMA traffic on the hot path: with a warm table_exists? memo, record! issues a single INSERT…ON CONFLICT DO NOTHING.
• Error handling: ActiveRecord::ActiveRecordError logs at DEBUG per attempt; any other StandardError logs at WARN once per process (latch serialized via a class-level mutex), then DEBUG.
• Idempotency, return values, and backfill!/all_names cache semantics are unchanged.
• Tests: adds an integration spec that poisons the pool schema cache and still registers the stream and asserts zero SCHEMA queries; integration harness bootstraps pgbus_stream_queues.

^{Written for commit cc3a2c3. Summary will update on new commits.}

[cubic-dev-ai]

All reported issues were addressed across 5 files

<sub>Reply with feedback, questions, or to request a fix.

Fix all with cubic | Re-trigger cubic</sub>

[cubic-dev-ai]

0 issues found across 2 files (changes from recent commits).

Confidence score: 5/5

• Automated review surfaced no issues in the provided summaries.
• No files require special attention.

_{Re-trigger cubic}