refactor!: make Client, Consumer, and Producer the public API - #164
Conversation
196c231 to
6742b88
Compare
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
https://github.com/efcasado/pulsar-elixir/blob/c4e5b0a6db84cfbf45449b04068f464318645025/lib/pulsar.ex#L180
Removing Pulsar.start_link/1 leaves the published docs/reader.md guide using {Pulsar, host: ..., name: ...} as a supervision-tree child at line 57. Pulsar is now only the application callback and has no matching child spec, so following that external-client example fails during supervisor initialization; replace it with {Pulsar.Client, ...} as in the updated module documentation.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Same as `start/1`, with the topic given positionally. | ||
| """ | ||
| @spec start(String.t(), keyword()) :: DynamicSupervisor.on_start_child() | ||
| def start(topic, opts) when is_binary(topic), do: start(Keyword.put(opts, :topic, topic)) |
There was a problem hiding this comment.
The previous public API allowed Pulsar.start_producer(topic) because its options defaulted to [], but the positional replacement requires an explicit second argument. Consequently, the direct migration Pulsar.Producer.start(topic) selects the keyword-list start/1 clause and raises FunctionClauseError; give opts the same empty-list default retained by Pulsar.Consumer.start/4.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4b16bec7d5
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b7dfe6f06e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
a27a449 to
376af97
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 376af9764e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d3c60631ce
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7e81ba9708
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6a14f23347
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| :initializing -> {:error, :not_ready} | ||
| {:ready, _shape} -> :ok |
There was a problem hiding this comment.
When metadata discovery succeeds but a producer worker is still registering—or subsequently receives a terminal response such as :IncompatibleSchema—this branch returns :ok solely from Discovery's topology status, even though the worker can immediately stop in Producer.Worker.handle_continue/2 and no usable producer exists. A caller relying on Pulsar.Producer.await_ready/2 can therefore proceed and have its first publish fail with :producer_died despite the successful barrier. Fresh evidence beyond the earlier bingo-example comment is that the same topology-only check is the public await_ready/2 implementation; readiness should include successful worker initialization or surface its failure.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bb50a9f1d7
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| try do | ||
| Supervisor.stop(client_name, :normal, timeout) | ||
| catch | ||
| :exit, _ -> :ok | ||
| :exit, _reason -> :ok | ||
| end |
There was a problem hiding this comment.
When applications create and stop dynamically named clients, init/1 stores each client's full broker configuration in :persistent_term, but this stop path no longer erases it. Every stopped client therefore permanently retains its options—including authentication and TLS data—and repeated transient clients steadily grow global memory; erase the entry before shutdown so a supervising parent can safely republish it during restart.
Useful? React with 👍 / 👎.
| case Topology.kind(consumer) do | ||
| :topology -> | ||
| client = Keyword.get(opts, :client, :default) | ||
| Topology.remove(consumer, Client.resource_supervisor(:consumers, client)) |
There was a problem hiding this comment.
When a producer root PID is accidentally passed to Pulsar.Consumer.stop/2, Topology.kind/1 identifies it only as a generic topology, and Topology.remove/2 falls back from the consumer supervisor to the PID's actual producer supervisor. The call therefore returns :ok after terminating a producer instead of the documented {:error, :not_found}; the inverse happens in Pulsar.Producer.stop/2, so PID operations must distinguish consumer and producer roots before removal.
Useful? React with 👍 / 👎.
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
1 participant
Summary
This PR makes
Pulsar.Client,Pulsar.Consumer,Pulsar.Producer, andPulsar.Readerthe library's public boundaries. Every broker connection, consumer, producer, and temporary reader consumer now lives below the client whose connection context it uses.It also replaces duplicated consumer and producer supervision with one shared topology, initializes topic metadata asynchronously, supports multiple isolated clients, and makes configuration explicit at the client or resource boundary.
This is a breaking change. Closes #163 and #165.
Motivation
Previously, the
:pulsarapplication owned clients and resources itself. That could run consumer callbacks before host dependencies were ready, block application startup on network metadata, discard resource startup failures, and left no clean way to supervise multiple independent clients. Consumers and producers also depended on client registries without living under that client.The host application now supervises
Pulsar.Client; the client owns its brokers, registries, consumers, and producers.Public API boundaries
Pulsar.ClientPulsar.ConsumerPulsar.ProducerPulsar.ReaderRuntime operations moved from
Pulsarto the module that owns them. The former consumer and producer GenServers are internal workers; public operations target a registered name or the stable root returned bystart.Pulsar.Client.consumers/1andPulsar.Client.producers/1therefore return one root per logical resource, independent of partition and worker count.Acknowledgement remains worker-specific because a message contains the broker-side consumer id that delivered it:
Ownership and lifecycle
Applications supervise one or more clients and may declare resources on each client:
:consumersor:producersoptionPulsar.Consumer.start/1orPulsar.Producer.start/1Pulsar.Reader.stream/2Starting establishes ownership and registration, not readiness. Metadata discovery and worker initialization continue asynchronously; callers may use
await_ready/2or handle{:error, :not_found}and{:error, :not_ready}. Consumer and producer branches recover independently, while broker infrastructure remains above both resource branches.Shared topology and discovery
The duplicated
ConsumerGroup,ProducerGroup,PartitionedConsumer, andPartitionedProducerimplementations are replaced by:Pulsar.Topology, the stable root for one logical resource;Pulsar.Topology.Discovery, which initializes and reconciles its shape;Pulsar.Topology.Resolver, which performs broker metadata and owner lookups;Pulsar.Topology.Group, which owns the workers for one topic or partition.A non-partitioned resource has one group; a partitioned resource has one group per partition. Discovery retries initial failures without blocking application startup, optionally polls for partition growth, and separately reconciles known groups without broker I/O. Terminal worker failures can therefore leave a stable but degraded root that a later local pass can recover.
Facade operations inspect topology groups rather than calling Discovery, so metadata I/O never serializes publishing or flow control. Producer routing retains unavailable partition slots and switches directly between complete partition widths during growth, avoiding temporary keyed remaps.
Reader
Pulsar.Readercreates a temporary non-durable consumer below an existing client and waits for its topology and workers.:startup_timeoutbounds that initialization separately from the stream's message inactivity:timeout.Configuration
Pulsar.ApplicationandPulsar.Configare removed. Starting the:pulsarapplication starts no clients or resources.Pulsar.ClientPulsar.ConsumerorPulsar.ProducerPulsar.Broker.OptionsOptions are validated at the public boundary and passed explicitly to owned processes.
Pre-existing bugs fixed
These bugs existed on
mainbefore this branch. They are listed separately so they do not disappear inside the architectural refactor.Stopping a runtime consumer or producer did not remove it. Calling
Supervisor.stop/1stopped a permanent child, so itsDynamicSupervisorimmediately restarted it. The public stop functions now terminate the child through its owner.A producer could disappear permanently after exhausting its restart intensity. Producer groups were registered as
:transient, so a group exiting with:shutdownafter exceeding its intensity was not restarted. Resource roots are now permanent children.The default startup delay defeated restart-intensity limits. The previous one-second delay plus jitter was wider than the restart-counting period, allowing a permanently failing resource to restart forever. Startup delay and jitter now default to zero and remain opt-in.
Duplicate declared resources were silently ignored. Startup results were discarded, so a duplicate registered name could be treated as successful. Client validation now rejects declarations that resolve to the same public name before startup.
Producers in the same group shared one topic epoch. Workers shared the group name while
Pulsar.Producer.EpochStorekeys epochs by producer name. Workers now receive distinct<group>-<n>names.Multiple clients could not use their documented child specs in one static tree. Every client used the same child-spec id. Client ids are now keyed by client name; consumer and producer ids are similarly keyed and namespaced.
Publishing through a stale producer pid could exit the caller. Tree inspection happened outside the exit-catching path. Publishing now returns
{:error, {:producer_died, reason}}.Partition growth could temporarily remap keyed messages through an intermediate modulus. Producers hashed using the number of partition groups already started, so a 4-to-6 expansion could briefly route with modulus 5 or leave that modulus in place after a partial start failure. Reconciliation now adds higher indexes first, and routing switches directly from the old contiguous width to the new one only after every intervening partition exists.
Publishing could treat
:restartingas a partition-group pid. Routing now ignores restarting children and returns{:error, :no_producers_available}while preserving partition selection.Operations against a missing or restarting client could exit instead of returning an error. Registry lookup, broker traversal, service discovery, and runtime resource startup now preserve the public error contracts while client branches are absent, shutting down, or restarting.
Stopping a supervised client could erase the replacement client's broker options.
Pulsar.Client.stop/2cleared per-client connection settings after the old process exited, racing a parent supervisor that had already restarted it. Cleanup now happens before shutdown, so the replacement always republishes the final value.Pulsar.Producer.EpochStore.get/4raised when the client's ETS table did not exist. Reads now return:error, matching the function specification and the write/delete APIs.Broker defaults differed depending on how the broker was started. Broker option validation and defaults now have one owner.
The reconnect ceiling disagreed with the documented default. Reconnect backoff now has one internal ceiling.
The manual-flow type contract allowed zero permits even though the implementation rejected them. Consumer flow-control specs and facade guards now consistently require a positive permit count.
Breaking changes and migration
~> 1.14~> 1.15(required for topology group auto-shutdown)config :pulsar, ...starts resources{Pulsar.Client, opts}Pulsar.start/1,start_link/1,start_client/1Pulsar.Clientor callPulsar.Client.start_link/1Pulsar.stop/1Pulsar.Client.stop/2Pulsar.start_broker/2,lookup_broker/2,stop_broker/2Pulsar.ClientfunctionsPulsar.start_consumer/4Pulsar.Consumer.start/4Pulsar.stop_consumer/2Pulsar.Consumer.stop/2Pulsar.lookup_consumer/2,Pulsar.get_consumers/2Pulsar.Consumer, or list logical roots withPulsar.Client.consumers/1Pulsar.ack/3,nack/3,send_flow/3Pulsar.ConsumerfunctionsPulsar.start_producer/2Pulsar.Producer.start/2Pulsar.stop_producer/2Pulsar.Producer.stop/2Pulsar.lookup_producer/2,Pulsar.get_producers/2Pulsar.Producer, or list logical roots withPulsar.Client.producers/1Pulsar.send/3Pulsar.Producer.send/3Reader.stream(topic, host: ...):clientAdditional migration notes:
Pulsar.Client.start_link/1and resourcestartcalls do not wait for readiness.Consumer.ack/2andConsumer.nack/2require the worker pid that received the message; a logical consumer name or root is ambiguous. Callback code can captureself()and pass it to asynchronous work.:host,:auth, or:socket_opts.<group>-<n>.Architecture guide
docs/architecture.mddescribes the registry-aware ownership tree, stable roots, asynchronous readiness, recovery model, and contributor implementation notes for reconciliation and routing.Validation
The unit and integration suites cover lifecycle, recovery, asynchronous discovery, routing, Reader behavior, and missing-process races. Formatting, Credo, Dialyzer, documentation, and doctests pass.
Follow-ups
The following remain outside this PR:
:killcan race the termination of its named descendants before the host supervisor restarts it. This predates the branch and does not occur during graceful shutdown or normal supervisor failure propagation.Pulsar.Producer.start/...andPulsar.Consumer.start/....