Go

The Go client takes a context.Context on every network call and separates building from executing: a builder chain is inert until you call Execute(ctx), Pop(ctx) or Get(ctx). It requires Go 1.24 and depends only on github.com/google/uuid and github.com/jackc/pgx/v5.

go get github.com/smartpricing/queen/clients/client-go

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

Construct a client

New accepts a string, a []string, a ClientConfig or a *ClientConfig, and returns an error rather than panicking on a bad URL.

client, err := queen.New("http://localhost:6632")
if err != nil {
    return err
}
defer client.Close(context.Background())

client, err := queen.New(queen.ClientConfig{
    URLs:                  []string{"http://broker-a:6632", "http://broker-b:6632"},
    BearerToken:           os.Getenv("QUEEN_TOKEN"),
    TimeoutMillis:         30000,
    LoadBalancingStrategy: "affinity",
    EnableFailover:        true,
})

Zero-valued numeric and string fields fall back to ClientDefaults, so a partially filled ClientConfig is safe.

Client options

Push

responses, err := client.Queue(queueName).Push(payload).Execute(ctx)
if err != nil {
	t.Fatalf("Failed to push message: %v", err)
}

Push accepts one payload, a []interface{} or a []map[string]interface{}. Chain .TransactionID(id) and .TraceID(id) before Execute.

responses, err := client.Queue("orders").
    Partition("acct-42").
    Push(map[string]interface{}{"id": 1}).
    TransactionID("order-1-created").
    Execute(ctx)

Execute returns []PushResponse, one entry per item in request order, carrying the broker’s status: queued, duplicate, buffered or failed. Supplying your own TransactionID is what makes a retry idempotent inside the queue’s dedup window.

Pop

messages, err := client.Queue("orders").
    Group("billing").
    Batch(10).
    Wait(true).
    TimeoutMillis(10000).
    Pop(ctx)

Pop returns []*Message and propagates errors, unlike the JavaScript, Python and C++ clients, which swallow a failed pop into an empty result.

.Batch(n) caps the messages returned. .Partitions(n) claims up to n partitions in one call, with Batch acting as a global cap across them and one shared LeaseID, so a single Renew extends them all. .AutoAck(true) on a pop asks the server to commit the cursor inside the pop transaction, which is at-most-once delivery.

Consume

err := client.Queue("jobs").
    Group("workers").
    Concurrency(4).
    Batch(1).
    Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
        return process(ctx, msg.Data)
    }).
    Execute(ctx)

There are two handler shapes. Consume takes a MessageHandler (func(context.Context, *Message) error) and is called once per message. ConsumeBatch takes a BatchMessageHandler (func(context.Context, []*Message) error) and is called once per popped batch, except that with Batch(1) and a single message returned, the worker calls the batch handler with a one-element slice through the single-message path. Execute(ctx) starts Concurrency goroutines and blocks until they all stop; Start(ctx) is an alias.

Cancel the ctx to stop every worker; the workers return ctx.Err(), which Execute filters out so a clean cancellation is not reported as a failure.

AutoAck is client-side. The worker never sends autoAck to the server: it calls POST /api/v1/ack after your handler returns, with status completed on nil and failed on an error. Because an ack is an offset commit, a nack clamps the group’s cursor at the failed message and everything after it in the popped batch is redelivered, so under .Each() the worker abandons the rest of the batch after a nack instead of producing duplicates and rejected acks.

Every delivered message carries a trace method that never panics:

_, _ = msg.Trace(ctx, queen.TraceConfig{
    TraceNames: []string{"tenant-acme"},
    EventType:  "step",
    Data:       map[string]interface{}{"stage": "charged"},
})

Acknowledge manually

responses, err := client.Ack(ctx, messages, true, queen.AckOptions{ConsumerGroup: "billing"})

Ack accepts *Message, Message, []*Message or []Message. success maps to completed or failed; AckOptions carries ConsumerGroup and Error (the failure reason recorded on the DLQ row if the nack exhausts the retry budget). One message goes to /api/v1/ack, several to /api/v1/ack/batch.

A rejected ack still arrives as HTTP 200 with success: false on the item, so check responses[i].Success. A nil error is not enough. The client compares the response length against the number of acks sent and errors out if they disagree, so a truncated response can never be attributed to the wrong message. ValidateMessages rejects an ack whose PartitionID is missing, because a TransactionID alone is not unique across partitions.

Transactions

_, err := client.Transaction().
    Queue("stage-2").Partition("acct-42").
    Push([]interface{}{map[string]interface{}{"value": 2}}).
    Ack(msg, queen.AckStatusCompleted, queen.AckOptions{ConsumerGroup: "billing"}).
    Commit(ctx)

Transaction() bundles pushes and acks into one PostgreSQL transaction, all or nothing. Acked messages contribute their LeaseID to requiredLeases, so the commit fails if a lease expired in the meantime.

Lease renewal

responses, err := client.Renew(ctx, messages)

Renew accepts a lease ID, a slice of lease IDs, or messages in any of the four shapes, and de-duplicates the IDs: a multi-partition pop shares one lease, so passing the whole slice issues one HTTP call.

Client-side buffering

_, err := client.Queue("events").
    Buffer(queen.BufferConfig{MessageCount: 500, TimeMillis: 200}).
    Push(payload).
    Execute(ctx)

With Buffer, Execute returns immediately with one PushResponse{Status: "buffered"} per item and the messages flush when the buffer reaches MessageCount (default 100) or TimeMillis (default 1000) elapses. Buffers are keyed per queue/partition.

client.Queue("events").FlushBuffer(ctx) flushes one address, client.FlushAllBuffers(ctx) flushes everything, and client.GetBufferStats() reports the active buffers, total buffered messages, oldest buffer age and flush count.

Queue configuration

_, err := client.Queue("orders").Config(queen.QueueConfig{
    LeaseTime:  300,
    RetryLimit: 5,
}).Create().Execute(ctx)

Create().Execute(ctx) sets configured: true on the returned map itself, for parity with the other SDKs. It is not the broker’s verdict: the broker’s own response is merged underneath it.

Admin

overview, err := client.Admin().GetOverview(ctx)
queues, err := client.Admin().ListQueues(ctx, queen.ListQueuesParams{Limit: 50})

client.Admin() covers the observability surface: resources, messages, traces, status and analytics, consumer groups, health, /metrics, /metrics/prometheus, the maintenance-mode switches, the DLQ listing, and the four per-queue analytics series. It is a superset of the JavaScript admin surface: PrometheusMetrics, ListDLQ, SeekConsumerGroupPartition, GetQueueLagAnalytics, GetQueueOpsAnalytics, GetQueueParkedReplicas and GetRetentionAnalytics have no JavaScript equivalent. The full route list with access levels is in Reference.

Dead-letter queue

dlq, err := client.Queue("orders").DLQ("billing").Limit(50).Get(ctx)

DLQ messages do carry Queue, RetryCount and ErrorMessage: the DLQ query builds them from queen.log_dlq, unlike a pop.

The broker does have a replay route that re-pushes a dead letter and drops its DLQ row, but this SDK does not reach it: Admin.RetryMessage returns an error without making a request. From Go, reprocess by reading the row and pushing the payload again, or call POST /api/v1/messages/{partitionId}/{transactionId}/retry directly. Either way the result is a new message at the tail of its partition, not the original one revived.

429 and 403

client, err := queen.New(queen.ClientConfig{
    URL:         "https://cell.example",
    BearerToken: token,
    Retry429:    &queen.Retry429Config{MaxAttempts: 20, BaseMs: 500, CapMs: 30000},
})

A 429 is retried in place against the same backend, honouring Retry-After (seconds) when present and otherwise backing off BaseMs * 2^attempt capped at CapMs, both with ±20 % jitter. MaxAttempts defaults to 10 for ordinary requests; a request marked with WithLongPollRetry() (which the pop and consume paths add automatically when Wait is true) retries unboundedly unless you set MaxAttempts, in which case it applies to both. A 429 never marks the backend unhealthy.

Errors from a non-2xx response are *HTTPError:

var httpErr *queen.HTTPError
if errors.As(err, &httpErr) {
    if httpErr.IsClusterSuspended() { return errStop }
    log.Printf("queen %d %s: %s", httpErr.StatusCode, httpErr.Code, httpErr.Error())
}

StatusCode, Code, Body and RetryAfterSeconds are all exposed, so you branch on the machine-readable code rather than on message text. A 403 is terminal: the consume worker returns it instead of retrying, because cluster_suspended, storage_quota_exceeded, feature_gated and forbidden cannot resolve themselves.

Logging

Set QUEEN_CLIENT_LOG to debug / true / 1, info, warn or error to enable the built-in logger; anything else (including unset) disables it. queen.SetLogLevel(...) changes it at runtime.

Shutdown

defer client.Close(context.Background())

The Go client installs no signal handlers. Close(ctx) flushes every buffer and closes the HTTP client, and returns an error if a flush failed, so a defer client.Close(ctx) in main plus your own signal.NotifyContext is the pattern. If you cancel the context you pass to Close, the final flush cannot complete and the buffered messages are lost.

Streaming

The windowed streaming SDK lives under clients/client-go/streams/ in the same module and binds to a *QueueBuilder through .AsStreamSource(). Its concepts (windows, watermarks, gates, state) are documented separately in Use Queen.