Expiration Policy

Overview

An expiration policy puts a hard time limit on how long a queue message may sit unconsumed. It solves a different problem than a dead-letter policy — this isn't about messages that fail processing, it's about messages that go stale: a price quote, a one-time code, a cache-invalidation signal, where late delivery is actively wrong, not just delayed. Instead of every consumer re-checking timestamps itself, the deadline lives on the message and the broker enforces it.

At the API level, SetExpirationSeconds(60) attaches a per-message TTL when you build the QueueMessage, and the clock starts the moment the broker accepts it via upstream.Send, not when a consumer picks it up. Let the TTL elapse unconsumed and the broker silently removes it — a later poll just comes back empty, no error, no trace.

Gotchas: expiration is silent — no DLQ routing, no event, just a message that vanishes — so pair it with monitoring if you need visibility into how much work is being dropped. The timer starts at send time, not when a consumer picks up the work, so a message can expire mid-backlog even while a consumer is actively polling. And setting the TTL too short for your real consumer lag just turns ordinary slowness into silent data loss.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: queues-stream/expiration-policy
//
// Demonstrates sending queue messages with an expiration (TTL) policy.
// Messages that are not consumed before the expiration time are automatically
// removed from the queue.
//
// Channel: go-queues-stream.expiration-policy
// Client ID: go-queues-stream-expiration-policy-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-queues-stream-expiration-policy-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues-stream.expiration-policy"

	// Send a message with 60-second expiration via upstream stream.
	upstream, err := client.QueueUpstream(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer upstream.Close()

	expiringMsg := kubemq.NewQueueMessage().
		SetChannel(channel).
		SetBody([]byte("expires in 60s")).
		SetExpirationSeconds(60)

	if err := upstream.Send("req-expiration", []*kubemq.QueueMessage{expiringMsg}); err != nil {
		log.Fatal(err)
	}

	// Drain the result.
	select {
	case res := <-upstream.Results:
		if res != nil && res.IsError {
			log.Printf("Error: %s", res.Error)
		} else {
			fmt.Println("Sent message with 60s expiration policy")
		}
	case <-time.After(3 * time.Second):
		fmt.Println("Sent message (no result confirmation within timeout)")
	}
}

How It Works

1. kubemq.NewQueueMessage().SetExpirationSeconds(60) sets a TTL on the message; after 60 seconds without being consumed, the broker silently discards it.
2. upstream.Send("req-expiration", []*kubemq.QueueMessage{expiringMsg}) delivers the TTL-bearing message via the upstream stream; the broker starts the expiration timer immediately upon receipt.
3. A consumer polling the queue after 60 seconds will find zero messages even though the send succeeded — this is expected TTL behaviour, not a bug.
4. Message expiration is useful for time-sensitive work items (e.g. cache-invalidation signals, short-lived alerts) where processing a stale message would be incorrect.

Related

• Pattern overview
• Go SDK Reference
• Send & Receive
• Ack All