feat(core): expose payload-offloading flag on handlerSpy (#464)

Author: kibertoad

README.md

@@ -676,6 +676,23 @@ const result = await myConsumer.handlerSpy.waitForMessageWithId('1')
 expect(result.processingResult).toEqual({ status: 'consumed' })
 ```
 
+When [payload offloading](#payload-offloading) is configured, the publisher's processing result
+exposes whether the message was offloaded to the external payload store. This lets you assert the
+offloading path was taken without mocking the store:
+
+```ts
+// Large message — payload was offloaded
+const offloaded = await myPublisher.handlerSpy.waitForMessageWithId('1', 'published')
+expect(offloaded.processingResult).toEqual({ status: 'published', offloaded: true })
+
+// Small message — sent inline, `offloaded` is omitted
+const inline = await myPublisher.handlerSpy.waitForMessageWithId('2', 'published')
+expect(inline.processingResult.offloaded).toBeUndefined()
+```
+
+The `offloaded` field is only present (set to `true`) when offloading actually happened, so existing
+exact-match assertions such as `toEqual({ status: 'published' })` keep passing for inline messages.
+
 ### Counters
 
 For load-testing scenarios where retaining all messages in the buffer may be too memory-intensive, handler spies provide lightweight counters that track how many messages were processed with each status:

packages/core/lib/types/MessageQueueTypes.ts

@@ -17,9 +17,16 @@ export type MessageProcessingResult =
       status: 'retryLater'
     }
   | {
-      status: 'consumed' | 'published'
+      status: 'consumed'
       skippedAsDuplicate?: boolean
     }
+  | {
+      status: 'published'
+      skippedAsDuplicate?: boolean
+      // Only set (to `true`) when the message payload was offloaded to the external
+      // payload store; omitted entirely for inline messages.
+      offloaded?: boolean
+    }
   | {
       status: 'error'
       errorReason: 'invalidMessage' | 'handlerError' | 'retryLaterExceeded'

packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts

@@ -4,6 +4,7 @@ import {
   type AsyncPublisher,
   type BarrierResult,
   DeduplicationRequesterEnum,
+  isOffloadedPayloadPointerPayload,
   type MessageInvalidFormatError,
   type MessageSchemaContainer,
   type MessageValidationError,
@@ -99,7 +100,9 @@ export abstract class AbstractPubSubPublisher<MessagePayloadType extends object>
       await this.sendMessage(maybeOffloadedPayloadMessage, options)
       this.handleMessageProcessed({
         message: parsedMessage,
-        processingResult: { status: 'published' },
+        processingResult: isOffloadedPayloadPointerPayload(maybeOffloadedPayloadMessage)
+          ? { status: 'published', offloaded: true }
+          : { status: 'published' },
         messageProcessingStartTimestamp,
         queueName: this.topicName,
       })

packages/sns/lib/sns/AbstractSnsPublisher.ts

@@ -6,6 +6,7 @@ import {
   type AsyncPublisher,
   type BarrierResult,
   DeduplicationRequesterEnum,
+  isOffloadedPayloadPointerPayload,
   type MessageInvalidFormatError,
   type MessageSchemaContainer,
   type MessageValidationError,
@@ -169,7 +170,9 @@ export abstract class AbstractSnsPublisher<MessagePayloadType extends object>
 
       this.handleMessageProcessed({
         message: parsedMessage,
-        processingResult: { status: 'published' },
+        processingResult: isOffloadedPayloadPointerPayload(payload)
+          ? { status: 'published', offloaded: true }
+          : { status: 'published' },
         messageProcessingStartTimestamp,
         queueName: topicName,
       })

packages/sns/test/publishers/SnsPermissionPublisher.payloadOffloading.spec.ts

@@ -118,9 +118,10 @@ describe('SnsPermissionPublisher - single-store payload offloading', () => {
 
       await publisher.publish(message)
 
-      await expect(
-        publisher.handlerSpy.waitForMessageWithId('1', 'published'),
-      ).resolves.toBeDefined()
+      // The handler spy exposes that the message took the offloading path, so tests can
+      // assert offloading happened without inspecting the store directly.
+      const spyResult = await publisher.handlerSpy.waitForMessageWithId('1', 'published')
+      expect(spyResult.processingResult).toEqual({ status: 'published', offloaded: true })
       await waitAndRetry(() => receivedSnsMessages.length > 0)
 
       // Check that the published message's body is a pointer to the offloaded payload.

packages/sqs/lib/sqs/AbstractSqsPublisher.ts

@@ -6,6 +6,7 @@ import {
   type AsyncPublisher,
   type BarrierResult,
   DeduplicationRequesterEnum,
+  isOffloadedPayloadPointerPayload,
   type MessageInvalidFormatError,
   type MessageSchemaContainer,
   type MessageValidationError,
@@ -160,7 +161,9 @@ export abstract class AbstractSqsPublisher<MessagePayloadType extends object>
       await this.sendMessage(payload, resolvedOptions, preBuiltBody)
       this.handleMessageProcessed({
         message: parsedMessage,
-        processingResult: { status: 'published' },
+        processingResult: isOffloadedPayloadPointerPayload(payload)
+          ? { status: 'published', offloaded: true }
+          : { status: 'published' },
         messageProcessingStartTimestamp,
         queueName: this.queue.name,
       })

packages/sqs/test/publishers/SqsPermissionPublisher.payloadOffloading.spec.ts

@@ -98,9 +98,10 @@ describe('SqsPermissionPublisher - payload offloading', () => {
 
       await publisher.publish(message)
 
-      await expect(
-        publisher.handlerSpy.waitForMessageWithId('1', 'published'),
-      ).resolves.toBeDefined()
+      // The handler spy exposes that the message took the offloading path, so tests can
+      // assert offloading happened without inspecting the store directly.
+      const spyResult = await publisher.handlerSpy.waitForMessageWithId('1', 'published')
+      expect(spyResult.processingResult).toEqual({ status: 'published', offloaded: true })
       await waitAndRetry(() => receivedSqsMessages.length > 0)
 
       // Check that the published message's body is a pointer to the offloaded payload.
@@ -140,5 +141,27 @@ describe('SqsPermissionPublisher - payload offloading', () => {
         getObjectContent(s3, s3BucketName, offloadedPayloadPointer),
       ).resolves.toBeDefined()
     })
+
+    it('does not flag small messages as offloaded', async () => {
+      const message = {
+        id: '2',
+        messageType: 'add',
+      } satisfies PERMISSIONS_ADD_MESSAGE_TYPE
+      expect(JSON.stringify(message).length).toBeLessThan(largeMessageSizeThreshold)
+
+      await publisher.publish(message)
+
+      const spyResult = await publisher.handlerSpy.waitForMessageWithId('2', 'published')
+      // `offloaded` is omitted entirely for inline messages, so the exact-match holds.
+      expect(spyResult.processingResult).toEqual({ status: 'published' })
+      await waitAndRetry(() => receivedSqsMessages.length > 0)
+
+      // Small messages are sent inline, not as an offloaded-payload pointer.
+      expect(receivedSqsMessages.length).toBe(1)
+      const parsedReceivedMessageBody = JSON.parse(receivedSqsMessages[0]!.Body!)
+      expect(parsedReceivedMessageBody.payloadRef).toBeUndefined()
+      expect(parsedReceivedMessageBody.offloadedPayloadPointer).toBeUndefined()
+      expect(parsedReceivedMessageBody.id).toBe('2')
+    })
   })
 })