Publish and consume messages

A message lets one part of an application ask another part to do work without waiting for that work to finish. The sender publishes a payload. A consumer receives it later and decides how to process it.

A successful publish means the messaging service accepted the message. It does not mean the worker finished the job. That distinction matters for retries, user-visible status, and operations that must not happen twice.

#Understand delivery and acknowledgement

#A queue distributes work; fanout distributes events

With competing consumers, a delivery goes to one worker in the group. This suits jobs that any available worker can perform. The examples use a queue with distribution: competition.

With fanout, separate consumers receive their own delivery stream. This suits an event that several independent parts of the application need to react to. The examples use a topic with distribution: fanout.

Neither model removes the need to tolerate repeated delivery. A worker can commit its business operation and then lose its acknowledgement. The messaging service may try again because it did not receive proof of completion.

#Why the acknowledgement comes last

An acknowledgement completes a delivery. Sending it before the business operation succeeds can lose work if the worker stops between those steps. Sending it afterwards still leaves a small gap: the business operation can succeed while the acknowledgement fails.

Make processing idempotent: repeating the same job should not repeat an already committed business effect. A durable job/event identifier and a transaction around recording completion and changing data can provide that behavior. The right mechanism depends on your business operation; Fluffy does not add it to your handler.

In the experiment, process a job, lose its acknowledgement, receive it again, and process again. Compare writing on every attempt with recognizing a completed job. Reset before comparing a different worker strategy.

#Pull and push change who starts the request

A pull worker calls Receive, processes the returned deliveries, then calls Ack or Nack using each delivery's token. A push consumer exposes a delivery endpoint and verifies the incoming request before processing it.

Pull workers need to remain running while waiting for messages. An HTTP service that scales to zero while idle cannot continuously poll a queue. Push processing must finish successfully before returning the response that completes the delivery.

#Configure a producer and worker

#Declare the queue and its edges

In fluffy-chainsaw.yaml, add a queue and attach existing API and worker runtimes:

queues:
  jobs:
    distribution: competition
runtimes:
  api:
    uses:
      queues:
        - name: jobs
          actions: [produce]
  worker:
    type: pull-worker
    consumes:
      - queue: jobs
        delivery: pull
        maximumMessages: 5
        wait: 10

Keep both runtimes' artifacts and other fields. Publishing belongs under uses; receiving belongs under consumes. There is no consume action to add to the producer grant.

#Supply delivery and runtime policy

The workspace chooses the product and the retry/retention policy. This excerpt requires an existing application provider and the rest of your normal runtime defaults:

resourceTypes:
  queues:
    reliable:
      product: gcp.pubsub.queue
  runtimes:
    pull-worker:
      product: gcp.cloud-run
      sizes:
        standard:
          cpu: 1
          memory: 512Mi
          minInstances: 1
          maxInstances: 1
          cpuIdle: false
environments:
  prod:
    queues:
      defaults:
        type: reliable
        provider: application
        delivery:
          ackDeadlineSeconds: 30
          maxProcessingSeconds: 300
          approximateMaxAttempts: 5
          minimumBackoffSeconds: 1
          maximumBackoffSeconds: 60
          sourceRetentionSeconds: 604800
          deadLetterRetentionSeconds: 604800

Select the worker's standard size through runtime policy. The worker type keeps a Cloud Run instance running with CPU available between requests. These are example capacity settings; choose capacity and concurrency for your workload.

Supply all seven delivery fields. Their values above are an example policy, not implied defaults. Retention bounds how long messages remain available. Retry backoff spaces out failed attempts. Dead-letter handling keeps repeatedly failing work separate from the ordinary delivery stream.

#Publish an event

Generate the API and worker packages separately, or generate all targets. For an application named example-app:

fluffy-chainsaw generate --application example-app --all --language typescript

Use the sidecar connection setup for your language. Imports assume application example-app; adjust them to your source location:

import { Metadata } from '@grpc/grpc-js'
import * as runtimev1 from './.fluffy/generated/runtimes/api/typescript/fluffy/runtime/example-app/api.js'
import * as messagingv1 from './.fluffy/generated/runtimes/api/typescript/fluffy/messaging/v1/messaging.js'

const client = new runtimev1.QueuesJobsClient(address, transport)
const published = await new Promise<messagingv1.PublishResponse>((resolve, reject) => {
  client.publish(
    messagingv1.PublishRequest.fromPartial({
      data: Buffer.from(JSON.stringify({ jobId: 'job-1' })),
      attributes: { type: 'report.requested' },
    }),
    new Metadata(),
    { deadline: Date.now() + 5000 },
    (error, value) => {
      if (error) reject(error)
      else resolve(value)
    },
  )
})

jobId is an application-defined identifier. Use a stable identifier appropriate to the job when implementing duplicate detection. Do not report the report as completed just because publication succeeded.

#Receive and acknowledge queue messages

The worker imports its own generated package, under .fluffy/generated/runtimes/worker/go, and its own fluffy/messaging/v1 types. This excerpt is one iteration of a processing loop in a function returning error:

import { Metadata } from '@grpc/grpc-js'
import * as runtimev1 from './.fluffy/generated/runtimes/worker/typescript/fluffy/runtime/example-app/worker.js'
import * as messagingv1 from './.fluffy/generated/runtimes/worker/typescript/fluffy/messaging/v1/messaging.js'

const client = new runtimev1.QueuesJobsClient(address, transport)
const batch = await new Promise<messagingv1.ReceiveResponse>((resolve, reject) => {
  client.receive(
    messagingv1.ReceiveRequest.fromPartial({ maximumMessages: 5, waitSeconds: 1 }),
    new Metadata(),
    { deadline: Date.now() + 5000 },
    (error, value) => {
      if (error) reject(error)
      else resolve(value)
    },
  )
})
for (const delivery of batch.deliveries) {
  try {
    await processJob(delivery.data)
  } catch (processingError) {
    await new Promise<void>((resolve, reject) => {
      client.nack(
        messagingv1.NackRequest.fromPartial({ token: delivery.token }),
        new Metadata(),
        { deadline: Date.now() + 5000 },
        (error) => (error ? reject(error) : resolve()),
      )
    })
    continue
  }
  await new Promise<void>((resolve, reject) => {
    client.ack(
      messagingv1.AckRequest.fromPartial({ token: delivery.token }),
      new Metadata(),
      { deadline: Date.now() + 5000 },
      (error) => (error ? reject(error) : resolve()),
    )
  })
}

process, processJob, or process_job is your application function (use the name in your selected language). It should return success only after committing the work or confirming that the same job already completed. The delivery token belongs to this delivery attempt; it is not the job's stable business identifier.

Use request deadlines and processing times consistent with your delivery policy. Handle an empty batch by continuing your normal polling loop; it does not prove the worker has failed.

#Configure a fanout topic

To notify independent consumers of an event, declare a topic and publish through it:

topics:
  events:
    distribution: fanout
runtimes:
  api:
    uses:
      topics:
        - name: events
          actions: [produce]
  worker:
    consumes:
      - topic: events
        delivery: push

Add these entries to the corresponding mappings. The generated producer uses NewTopicsEventsClient(connection).Publish. In the workspace, define a topic type with product: gcp.pubsub.topic and select it under the environment's topics.defaults, with a complete delivery policy just as for the queue above.

#Verify push delivery

For the events topic, implement POST /.well-known/fluffy/v1/topics/events/deliveries. Pass the original method, path, headers, and body to NewInboundMessagingClient(connection).VerifyPushRequest, then process the verified delivery. An arbitrary POST to that path is not a trusted message.

Do not acknowledge through an early HTTP success while processing continues in the background. Return success only after successful processing; retry and dead-letter policy handles subsequent attempts when processing fails.

#Run and verify

Run locally, publish a message with a recognizable development job ID, and inspect the worker's result. Force processing to fail before acknowledgement and confirm another attempt can occur. Then test a repeated job ID and confirm the business operation is not duplicated.

For push delivery, also send an unverified request to your development endpoint and confirm it is rejected before business processing. Diagnose verification failures instead of disabling verification to make a test request pass.

#Reference: consumer and delivery constraints

Duplicate consume entries for the same source and dead-letter depth within one runtime are rejected. A dead-letter consumer must be part of the declared, resolved delivery topology; changing depth does not mean replaying a message has succeeded.

#Delivery policy ranges

All of these fields must be supplied in the selected environment's resolved delivery policy:

The attempt limit is explicitly approximate. Do not build business correctness around an exact number of deliveries or assume a retry happens at an exact wall-clock time.

#Diagnose an idle or failing worker

Continue with Connect services for calls where one runtime needs a response from another during the request.