|
| 1 | +// Copyright (c) 2025 Uber Technologies, Inc. |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +// Package publish sends a message to the queue behind a topic key. It owns the |
| 16 | +// lookup-and-send plumbing every pipeline stage otherwise repeats — resolve the |
| 17 | +// key to a queue and a topic name, wrap the payload in a message, publish — and |
| 18 | +// the message-ID convention that controls deduplication (see IntentID). |
| 19 | +// |
| 20 | +// Every producer publishes through this package. Building a message anywhere |
| 21 | +// else would put the ID choice back at each call site, which is the mistake the |
| 22 | +// convention exists to prevent, so a linter restricts message construction to |
| 23 | +// here and to the queue backends. |
| 24 | +package publish |
| 25 | + |
| 26 | +import ( |
| 27 | + "context" |
| 28 | + "fmt" |
| 29 | + "strings" |
| 30 | + "sync/atomic" |
| 31 | + "time" |
| 32 | + |
| 33 | + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" |
| 34 | + "github.com/uber/submitqueue/platform/consumer" |
| 35 | +) |
| 36 | + |
| 37 | +// Message publishes payload to the topic registered for key. |
| 38 | +// |
| 39 | +// msgID selects the dedup behavior, so the caller must choose it deliberately. |
| 40 | +// The queue deduplicates on (topic, partition key, message ID) against every |
| 41 | +// row it has not garbage-collected yet, consumed ones included — a window with |
| 42 | +// no upper bound on a busy partition. A publish that collides is reported as a |
| 43 | +// success and writes nothing, and nothing retries it. |
| 44 | +// |
| 45 | +// Build msgID with IntentID: name the entity the message is about and the cause |
| 46 | +// this particular message exists for. A retry of the same cause then dedups, |
| 47 | +// which is what makes redelivery safe, while a new cause about the same entity |
| 48 | +// can never be swallowed by an older row. |
| 49 | +func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error { |
| 50 | + q, ok := registry.Queue(key) |
| 51 | + if !ok { |
| 52 | + return fmt.Errorf("no queue registered for topic key %s", key) |
| 53 | + } |
| 54 | + topicName, ok := registry.TopicName(key) |
| 55 | + if !ok { |
| 56 | + return fmt.Errorf("no topic name registered for topic key %s", key) |
| 57 | + } |
| 58 | + |
| 59 | + msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) |
| 60 | + return q.Publisher().Publish(ctx, topicName, msg) |
| 61 | +} |
| 62 | + |
| 63 | +// IntentID names the occasion to publish rather than the entity published |
| 64 | +// about: entityID says what the message concerns, and cause says why this |
| 65 | +// particular message exists. |
| 66 | +// |
| 67 | +// Passing no cause asks for at-most-once delivery per entity — every later |
| 68 | +// publish about that entity is dropped while an earlier row survives. That is |
| 69 | +// right only for a hand-off that happens once in an entity's life, such as |
| 70 | +// announcing that it was created. Anything re-sent by design — a wake-up, a |
| 71 | +// poll, a re-dispatch, a dead-letter reconciliation — must name its cause, or |
| 72 | +// it collides with that one-shot publish and is lost. |
| 73 | +// |
| 74 | +// Each cause segment must be stable across redeliveries of one occurrence and |
| 75 | +// different between occurrences, so derive it from whatever provoked the |
| 76 | +// publish: the dependency that reached a terminal state, the build and the |
| 77 | +// status observed, the dead letter being reconciled. A wall-clock reading or a |
| 78 | +// random value satisfies "different" while destroying "stable", leaving every |
| 79 | +// redelivery to publish again. An empty segment carries no information and |
| 80 | +// makes two different occasions share an ID, so callers pass none. |
| 81 | +func IntentID(entityID string, cause ...string) string { |
| 82 | + if len(cause) == 0 { |
| 83 | + return entityID |
| 84 | + } |
| 85 | + return entityID + "/" + strings.Join(cause, "/") |
| 86 | +} |
| 87 | + |
| 88 | +// sequence breaks ties between UniqueID calls that land on the same clock |
| 89 | +// tick: some platforms quantize time.Now coarsely enough for consecutive calls |
| 90 | +// to read the same nanosecond. |
| 91 | +var sequence atomic.Uint64 |
| 92 | + |
| 93 | +// UniqueID returns a message ID no earlier publish has used, so the publish |
| 94 | +// cannot be deduplicated away. |
| 95 | +// |
| 96 | +// This is the fallback for a cause with nothing stable to name it by, and it |
| 97 | +// costs the idempotency IntentID preserves: a redelivery mints a fresh ID and |
| 98 | +// publishes a second time, so the consumer has to absorb the duplicate. Prefer |
| 99 | +// IntentID wherever the cause can be identified. |
| 100 | +func UniqueID(id string) string { |
| 101 | + return fmt.Sprintf("%s@%d-%d", id, time.Now().UnixNano(), sequence.Add(1)) |
| 102 | +} |
0 commit comments