Durably commits a consumer group's position for topic from cursor (a token from fetch/ fetch_group). Returns :ok, or {:error, :invalid_cursor} for a bad token.

Creates topic. The client does not specify partitions: the keyspace is internal.

Encodes internal consume positions into an opaque cursor. Public because the streaming push path (the connection forwarding {:log_records, ...}) must turn the pushed positions into a client cursor, the same token fetch/fetch_group return.

Fetches up to max records per current range of topic from the position in cursor (nil or :start begins at the beginning). Returns {:ok, records, next_cursor}, advance by passing next_cursor back. Records carry no client-visible offset.

Fetches for a consumer group, resuming from the group's durably committed position (or the beginning if it never committed). Returns {:ok, records, next_cursor}; the client processes the records and then commit/4s next_cursor to advance the durable position (at-least-once).

Fetches for member of consumer group on topic: consults the coordinator for the member's assigned ranges (registering it if new: the fetch is the heartbeat) and consumes only those ranges from the group's committed positions. Returns {:ok, records, next_cursor}; the opaque cursor covers only the member's slice, so commit/4 advances just its ranges (the client never sees a range id). Members of the same group thus consume disjoint records in parallel.

Produces records (each %{"value" => binary, optional "key" => binary, optional "headers" => map}) to topic, routed by key. Returns {:ok, produced_count}; the client gets no offsets.

Appends already-built Malachi.Log.Records (offset unassigned) to topic: the binary protocol path, which decodes records off the wire directly, skipping the JSON map→record step of produce/3. Returns {:ok, count} or {:error, reason}.

Acks count streamed records for group at cursor: durably commits the position (at-least-once) and returns count records of window credit, unblocking further pushes. Returns :ok, or {:error, :invalid_cursor} for a bad token.

Like stream_ack/5 but for a group member: re-polls the coordinator (a heartbeat that also refreshes the member's ranges, so a rebalance is picked up on the next ack) before acking. Returns :ok or {:error, :invalid_cursor}.

Opens a streaming subscription: registers the calling process as a push subscriber of topic for consumer group, bounded by a credit window (max in-flight records) and a max per-push batch. The broker resumes from the group's committed position, pushes an initial backlog, and then pushes new records on produce as {:log_records, topic, records, positions} messages to the caller. Returns :ok.

Like subscribe/5 but for a consumer-group member: registers the member with the coordinator, scopes the push stream to its assigned ranges (server-side: the client still only gets records + an opaque cursor), and lets the broker leave the group when the calling process exits. The member stays alive by acking (stream_ack_member/7); the coordinator does not run in the broker, so the broker never calls it (deadlock-safe).