Method tick

packages/cli/src/util/dev/queue-broker.ts:489–550

()

  }

  private tick(): void {
    const now = Date.now();

    for (const [key, record] of this.idempotencyRecords) {
      if (record.expiresAt <= now) {
        this.idempotencyRecords.delete(key);
      }
    }

    for (const [messageId, record] of this.duplicateMessages) {
      if (record.expiresAt <= now) {
        this.duplicateMessages.delete(messageId);
      }
    }

    for (const group of this.consumerGroups) {
      const groupDeliveries = this.deliveryState.get(group.id);
      if (!groupDeliveries) continue;

      for (const [messageId, state] of groupDeliveries) {
        const message = this.messages.get(messageId);

        // Clean up if message expired
        if (message) {
          const expiresAt =
            new Date(message.createdAt).getTime() + message.retentionMs;
          if (expiresAt < now) {
            groupDeliveries.delete(messageId);
            this.maybeCleanupMessage(messageId);
            continue;
          }
        }

        if (!message) {
          groupDeliveries.delete(messageId);
          continue;
        }

        if (state.status === 'in-flight' && state.leaseExpiresAt < now) {
          if (state.deliveryCount >= group.maxDeliveries) {
            output.debug(
              `queues: message ${messageId} exceeded maxDeliveries (${group.maxDeliveries}) for group "${group.name}", dropping`
            );
            groupDeliveries.delete(messageId);
            this.maybeCleanupMessage(messageId);
          } else {
            state.status = 'pending';
            state.visibleAt = now + group.retryAfterMs;
            state.leaseExpiresAt = 0;
          }
          continue;
        }

        // Dispatch pending messages that are now visible
        if (state.status === 'pending' && state.visibleAt <= now) {
          this.dispatchToConsumer(message, group).catch(err => {
            output.debug(`queues: unexpected dispatch error: ${err}`);
          });