Go client

Import path github.com/smartpricing/queen/clients/client-go, package name queen. Requires Go 1.24 or newer. Every network method takes a context.Context as its first argument. Version 1.3.0, aligned with the broker’s 1.3 line.

import queen "github.com/smartpricing/queen/clients/client-go"

The streaming SDK is a separate package under .../client-go/streams.

Constructor

func New(config interface{}) (*Queen, error)

config accepts a string (single URL), a []string (multiple URLs), a ClientConfig, or a *ClientConfig. Anything else returns invalid config type: expected string, []string, or ClientConfig.

ClientConfig

The exported ClientDefaults, QueueDefaults, ConsumeDefaults, PopDefaults and BufferDefaults variables hold these values, so an application can read one instead of restating it. Unlike the other clients, the load balancer is constructed unconditionally, even for a single URL, so LoadBalancingStrategy and the health settings are always in play.

Retry429Config

Retry429Config (MaxAttempts, BaseMs, CapMs) covers HTTP 429 separately from RetryAttempts, and a zero field means “use the kind-based default”; the defaults, the Retry-After contract and the jitter are in What the clients do with a 429.

*Queen

Ack

messages accepts *Message, Message, []*Message or []Message; anything else returns an error. success maps to AckStatusCompleted ("completed") or AckStatusFailed ("failed"). AckOptions carries ConsumerGroup and Error.

One message uses POST /api/v1/ack, more than one uses POST /api/v1/ack/batch. Both return one AckResponse{Success, Error} per message in request order. ValidateMessages runs first, so a message missing TransactionID or PartitionID fails before any request is sent.

Renew

Accepts a lease-id string, []string, or any of the message forms. Lease ids are deduplicated first: with multi-partition pop every message in a batch shares one LeaseID and one extend call renews every claimed partition, so passing the whole slice issues one HTTP call. RenewResponse carries LeaseID, Success, NewExpiresAt time.Time and Error.

*QueueBuilder

client.Queue(name). Configuration methods return *QueueBuilder; terminal methods are Push, Pop, Consume, ConsumeBatch, Create, Delete, DLQ and FlushBuffer.

Addressing

Queue lifecycle

*OperationBuilder has one method: Execute(ctx) (map[string]interface{}, error). Create validates the queue name first and adds "configured": true to the returned map for parity with the other clients.

QueueConfig holds the fields Create can send, each mapped onto its camelCase wire key: LeaseTime, RetryLimit, DelayedProcessing, WindowBuffer, RetentionSeconds, CompletedRetentionSeconds and EncryptionEnabled, whose QueueDefaults values are the shared client-side defaults listed under Queue option defaults, plus three booleans the other clients’ defaults dicts do not carry: RetentionEnabled, DeadLetterQueue and DlqAfterMaxRetries, all false, which by the rule below means they are not sent at all.

Two builder methods sit on *OperationBuilder for this:

_, err := client.Queue("orders").
	Config(queen.QueueConfig{LeaseTime: 60}).
	Option("deadLetterQueue", false). // an explicit false, unsendable from QueueConfig
	Option("retentionSinkHold", nil). // null: back to the default
	Create().
	Execute(ctx)

A broker older than 1.6.0 ignores mode and replaces either way, which is what it has always done.

Producing

*PushBuilder:

PushResponse carries Status ("queued", "duplicate", "failed"), TransactionID and Error. The client mints a UUIDv7 transactionId for items that have none.

res, err := client.Queue("orders").
	Partition("customer-42").
	Push(map[string]any{"orderId": 9137, "amount": 99.5}).
	Execute(ctx)
if err != nil {
	return err
}
// res[0].Status == "queued"

Consuming

Pop

func (qb *QueueBuilder) Pop(ctx context.Context) ([]*Message, error)

Returns queue name, namespace, or task is required when none of the three is set. Route selection: queue plus non-Default partition uses /api/v1/pop/queue/:queue/partition/:partition; queue alone uses /api/v1/pop/queue/:queue; namespace or task alone uses /api/v1/pop.

messages, err := client.Queue("orders").
	Batch(10).
	Wait(true).
	Pop(ctx)
if err != nil {
	return err
}

PopResult

func (qb *QueueBuilder) PopResult(ctx context.Context) (PopResult, error)

The same call as Pop, returning PopResult{Messages, Autopilot}: the broker’s account of how it sized the claim, with Partitions, Batch, and an optional WaitMillis pacing hint the consume loop honours in place of its own delay between empty polls. Autopilot is nil when the pop did not engage autopilot, when the broker is older than 1.2, or when the answer was a bodiless 204, which has no body to carry an echo. See pop autopilot.

Consume and ConsumeBatch

func (qb *QueueBuilder) Consume(ctx, handler MessageHandler) *ConsumeBuilder
func (qb *QueueBuilder) ConsumeBatch(ctx, handler BatchMessageHandler) *ConsumeBuilder

err = client.Queue("orders").
	Group("billing").
	SubscriptionMode("all").
	Limit(1).
	Each().
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		fmt.Println(msg.Data)
		return nil
	}).
	Execute(ctx)
if err != nil {
	return err
}

MessageHandler is func(ctx context.Context, msg *Message) error; BatchMessageHandler is func(ctx context.Context, msgs []*Message) error. *ConsumeBuilder exposes Execute(ctx) error and Start(ctx) error (Start is an alias). Both block until every worker stops; cancel the context to stop them.

Worker loop behaviour:

• A handler error with AutoAck on nacks the message and the worker continues.
• A 429 that escapes the HTTP layer’s own retry backs off and continues.
• A 403 stops the worker. Use (*HTTPError).IsClusterSuspended() to distinguish the terminal case.

(*Message).Trace

func (m *Message) Trace(ctx context.Context, config TraceConfig) (*TraceResponse, error)

TraceConfig carries TraceName string, TraceNames []string, EventType string and Data map[string]interface{}. Posts to /api/v1/traces.

Buffering

FlushBuffer(ctx) error flushes this builder’s queue/partition buffer. *BufferManager exposes Add, Flush(ctx, key), FlushAll(ctx), GetStats() BufferStats, GetBuffer(key) and Clear(). BufferStats carries ActiveBuffers, TotalBufferedMessages, OldestBufferAge and FlushesPerformed.

The buffer is bounded and lossless under errors. At MaxSize waiting messages (default 4 * MessageCount), Add blocks, honoring its context, until the flusher drains below the bound. A batch whose POST fails goes back to the front of the buffer, in order, and is retried every RetryDelayMillis; it is never dropped, though occupancy can overshoot MaxSize by that one restored batch. A broker outage therefore shows up as blocked producers and a full buffer, not as silent loss.

Dead-letter queue

DLQ(consumerGroup string) *DLQBuilder.

The listing is read-only, but Admin.RetryMessage(ctx, partitionID, transactionID) replays. Since broker 1.6.0 it runs on a move: one transaction claims the newest dead-letter row at the address under a row lock, pushes the snapshot back into the log and deletes that row. Only dead-lettered addresses replay, and a live one answers 404.

It is now idempotent by row. The replayed frame carries the deterministic transaction id dlq:<dead-letter row id>, a second call for the same address answers 404 because the row is gone, and two concurrent callers serialise on the lock so the loser is told the row is gone rather than pushing a second copy. Only the addressed consumer group’s record is removed: an address can carry one row per group, and the others stay where they were. A result:"duplicate" verdict means nothing was written and nothing was removed, and a 503 means push maintenance is on, which refuses the move and leaves the row. The client still sends it with failover retry disabled, because it is a write and a blind resend by the transport would hide a verdict the caller has to read. Admin.MoveMessageToDLQ is still a stub, because the broker registers no force-move route.

*TransactionBuilder

client.Transaction(). Pushes and acks in one PostgreSQL transaction, all-or-nothing.

messages, err := client.Queue("orders").
	Group("invoicing").
	SubscriptionMode("all").
	Batch(1).
	Wait(true).
	Pop(ctx)
if err != nil {
	return err
}
message := messages[0]

_, err = client.Transaction().
	Queue("invoices").
	Push(map[string]any{"orderId": message.Data["orderId"], "invoiced": true}).
	Ack(message, queen.AckStatusCompleted, queen.AckOptions{ConsumerGroup: "invoicing"}).
	Commit(ctx)
if err != nil {
	return err
}

The two operations can be given in either order: the builder collects them and Commit sends one request.

A transaction is atomic, not exactly-once end to end; see the limit.

*Admin

client.Admin(). Every method takes a context and returns map[string]interface{}, the broker’s JSON body verbatim, except where noted.

Resources

ListQueuesParams carries Namespace, Task, Limit, Offset. GetPartitions requires a queue name and returns the queue-detail payload: the broker bundles partition data there rather than exposing a partitions collection.

Messages, DLQ and traces

ListMessagesParams carries Queue, Partition, Status, ConsumerGroup, Limit, Offset. ListDLQParams carries Queue, ConsumerGroup, Partition, From, To, Limit, Offset.

Status and analytics

GetStatusParams carries Queue, Namespace, Task.

Consumer groups

SeekConsumerGroupOptions carries Timestamp string (RFC3339) and ToEnd bool, which are mutually exclusive. The wire body is {"toEnd": true} or {"timestamp": "…"}.

System

Errors

Non-2xx responses come back as *HTTPError:

var he *queen.HTTPError
if errors.As(err, &he) && he.StatusCode == 429 { /* ... */ }

IsClusterSuspended() reports the terminal 403 that no amount of retrying resolves. Error() renders as HTTP <status> [<code>]: <body>.

A 204 response yields a nil map: a 204 carries no body at all.

*HttpClient

Reachable via GetHttpClient() when you need a route the SDK does not wrap.

All four return (map[string]interface{}, error); a top-level JSON array is wrapped as {"data": [...]}. WithLongPollRetry() is the only RequestOption: it marks the request as a long-poll pop so a 429 backs off indefinitely instead of using the bounded budget.

Logging

Set by the QUEEN_CLIENT_LOG environment variable, read once at package init:

SetLogLevel(level LogLevel) and GetLogLevel() LogLevel change it at runtime.

Helpers

Message

Streaming SDK

import "github.com/smartpricing/queen/clients/client-go/streams"

s := streams.From(queen.Queue("orders").AsStreamSource())

AsStreamSource() on a *QueueBuilder adapts it to runtime.Source. Every combinator returns a new *Stream.

Every WindowX constructor takes the same five WindowOption values, which control when a window closes and what happens to whatever arrives after it did:

To(sink) reuses the source partition name for each push; ToPartitioned(sink, resolver) chooses the destination partition instead, resolver being a fixed partition string or a func(value interface{}) string called per emit (anything else panics), the mirror of the JavaScript .to(queue, { partition }) form. The resolver is not part of the config hash, so switching between the two, or changing the resolver, neither trips the registration mismatch check nor resets query state.

streams/helpers provides TokenBucketGate(TokenBucketGateOptions) and SlidingWindowGate(SlidingWindowGateOptions), both returning a GateFn.

RunOptions

*Runner exposes Metrics() Metrics with the counters CyclesTotal, FlushCyclesTotal, MessagesTotal, PushItemsTotal, StateOpsTotal, LateEventsTotal, ErrorsTotal, GateAllowsTotal, GateDenialsTotal and LastError.

The chain shape is fingerprinted into a config hash; re-deploying a different chain under the same QueryID is rejected at registration unless Reset is set. Only operator kinds and their structural config are hashed, not the bodies of your functions.

Same surface, other languages

Parity here is a claim about the surface, not about shared code: each client is written natively in its own idiom. These sixteen rows are that surface, read out of clients/ on this tree. Six of them are full parity across all six clients; the other ten are where a client stops.

The ten divergent rows, with their conditions:

• Key/value. The seven operations are get, getMany, getPrefix, put, putIfAbsent, delete and incr. C++ wraps five of them: getMany and getPrefix have no method, and both remain reachable through POST /api/v1/kv on the shared HttpClient. getPrefix inside a transaction is refused by the broker in every client, which is its rule and not a client’s gap.
• Timers. C++ has schedule and cancel only; peek and list are reads and stay on GET /api/v1/timers/{queue}[/{timerKey}]. There is no reschedule operation anywhere, because schedule is the upsert and status reports which happened; Python, Go, Rust and PHP still spell a reschedule alias over it, JavaScript and C++ do not.
• once. JavaScript (kv.once, transaction().once) and Python (kv.once, transaction().once) fold putIfAbsent plus required into the question people actually ask, “did I win?”. The other four write the putIfAbsent themselves, which is the same wire and one line longer.
• kv on a streams operator context. No client has it, so it is not a row. Inside a stream the state primitive is state_ops, which commits with the sink and the ack in the cycle’s own transaction; a key/value write from an operator would not, and that atomicity is the thing the stream already gives you for free.
• Admin facade. C++ has no Admin class. Every management and observability route is still reachable, through the shared HttpClient returned by get_http_client().
• affinity. The C++ LoadBalancer implements round-robin and "session" only, and load_balancing_strategy = "affinity" falls through to round-robin silently. The strategy matters because it keeps one consumer’s pops on one backend, which is what keeps two clients from contending on the same partition claim.
• Streams. The Stream builder, the four window kinds, gates and the /streams/v1/* runtime are in the JavaScript, Python, Go and Rust clients. PHP and C++ have none of it.
• Lease renewal. In C++, renew_lease() sets a flag the consumer loop does not act on: the worker carries an unimplemented placeholder where the timer would go. Call client.renew(message) yourself before the lease expires.
• Per-message trace. Where it exists it is attached by the consume loop, so a message returned by a manual pop() never carries it. Rust records a trace through Admin::record_trace instead of hanging a method off the message. C++ cannot hang one off a JSON object at all, and its trace hook is an explicit no-op.
• Host override. JavaScript (hostHeader) and Rust (host_header) advertise a request authority, and a TLS SNI, independent of the address the socket dials. That is what addresses a named tenant cluster behind a shared proxy endpoint. The other four clients send the dialled address.
• pop failures. Go, Rust and PHP surface a 4xx, an exhausted 429 budget, a terminal 403 or a network fault to the caller, so an empty result means an empty queue. JavaScript, Python and C++ log the failure and return an empty result, which does not distinguish an empty queue from revoked credentials.

The go suite in the test matrix runs the root-package unit tests, then ./tests/ and ./tests/streams_integration/ with -count=1 so a cached result can never hide a live-broker failure, on the single, ha and tenanted topologies. The ./streams/... package is green at 13 of 13 on this tree, with one known deterministic failure named on that page.