diff --git a/platform/consumer/README.md b/platform/consumer/README.md index 8c0507ff..ba6ea576 100644 --- a/platform/consumer/README.md +++ b/platform/consumer/README.md @@ -133,7 +133,6 @@ Several mechanisms can delay work; they mean different things. Pick by what you' | "I'm still working — keep my lease" | `delivery.ExtendVisibilityTimeout(...)` | blocked behind the in-flight delivery | | "Done for now — wake this partition in N ms" | `delivery.Hold(N)` then `return nil` | paused behind the postponed message (barrier), redelivers first in order | | "Stop this controller/partition from outside" (tests, operators) | consumer gate (`platform/extension/consumergate`) | parked in flight until the gate opens | -| "Defer *other* work" — a delayed message to another topic or key | `Publisher.PublishAfter` | not involved — it's a fresh publish | Gate vs hold, since both pause a partition: the **gate** is an external, event-ended stop — someone stops the controller at the door, before `Process` ever sees the message. **Hold** is a controller-chosen, timer-ended wait — the controller saw the work and decided to come back later. Business logic never closes or opens gates; a controller that needs to back off uses hold. diff --git a/platform/extension/messagequeue/README.md b/platform/extension/messagequeue/README.md index c3bc9b2b..6f221d78 100644 --- a/platform/extension/messagequeue/README.md +++ b/platform/extension/messagequeue/README.md @@ -13,20 +13,12 @@ Publishes messages to topics. ```go type Publisher interface { Publish(ctx context.Context, topic string, message entityqueue.Message) error - PublishAfter(ctx context.Context, topic string, message entityqueue.Message, delayMs int64) error Close() error } ``` (`entityqueue` is `github.com/uber/submitqueue/platform/base/messagequeue`.) -**`PublishAfter`** inserts a fresh message that becomes visible to subscribers only after `delayMs`. It is distinct from `Nack(requeueAfterMillis)` even though both can produce "next delivery happens at T+delay": - -- `Nack` is "this delivery failed, try again" — it bumps `retry_count` and eventually trips DLQ. -- `PublishAfter` is "postpone this work" — `retry_count` resets to 0, DLQ stays available for true failures. - -For a consumer deferring its *own current delivery* ("check back in N ms"), prefer `Delivery.Postpone` (below) over ack-plus-`PublishAfter`: it needs no publisher, no fresh message id, and keeps the same log row. `PublishAfter` remains the tool for deferring *other* work — publishing a delayed message to a different topic or key. Use `Nack` for processing failures. - ### Subscriber Consumes messages from topics with per-subscription configuration. diff --git a/platform/extension/messagequeue/mock/publisher_mock.go b/platform/extension/messagequeue/mock/publisher_mock.go index c313b7bd..593aa639 100644 --- a/platform/extension/messagequeue/mock/publisher_mock.go +++ b/platform/extension/messagequeue/mock/publisher_mock.go @@ -68,17 +68,3 @@ func (mr *MockPublisherMockRecorder) Publish(ctx, topic, message any) *gomock.Ca mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Publish", reflect.TypeOf((*MockPublisher)(nil).Publish), ctx, topic, message) } - -// PublishAfter mocks base method. -func (m *MockPublisher) PublishAfter(ctx context.Context, topic string, message messagequeue.Message, delayMs int64) error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "PublishAfter", ctx, topic, message, delayMs) - ret0, _ := ret[0].(error) - return ret0 -} - -// PublishAfter indicates an expected call of PublishAfter. -func (mr *MockPublisherMockRecorder) PublishAfter(ctx, topic, message, delayMs any) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PublishAfter", reflect.TypeOf((*MockPublisher)(nil).PublishAfter), ctx, topic, message, delayMs) -} diff --git a/platform/extension/messagequeue/mysql/README.md b/platform/extension/messagequeue/mysql/README.md index 9c8d7709..11ca8bc3 100644 --- a/platform/extension/messagequeue/mysql/README.md +++ b/platform/extension/messagequeue/mysql/README.md @@ -112,8 +112,6 @@ platform/extension/messagequeue/mysql/ | `queue_partition_leases` | Partition lease coordination | `(consumer_group, topic, partition_key)` | | `queue_subscriber_heartbeats` | Active subscriber tracking | `(consumer_group, topic, subscriber_name)` | -`queue_messages` has a `visible_after BIGINT UNSIGNED NOT NULL DEFAULT 0` column that supports `Publisher.PublishAfter`: subscribers' `FetchByOffset` skips rows where `visible_after > now`. Default 0 means immediately visible, so existing rows continue to behave as before — the column is back-compatible. - `queue_delivery_state` has a `postponed BOOLEAN NOT NULL DEFAULT FALSE` column that supports `Delivery.Postpone`. `MarkPostponed` sets `invisible_until = now + delay`, resets `retry_count` to 0, and sets the flag. While the flag is set and the row is invisible, the poll loop treats the message as a **barrier** — it stops scanning the partition instead of skipping past it (nacked rows keep skip-and-continue semantics, so a failed message never halts its partition). On the next `MarkDelivered` the flag is consumed: the `retry_count` increment is skipped and the flag cleared, so a postponed redelivery restarts as attempt 1 and only consecutive real failures count toward `Retry.MaxAttempts`. Default FALSE keeps existing rows back-compatible. See `schema/` for full SQL definitions. See the [RFC](../../../doc/rfc/sql-queue-rfc.md#database-schema) for field-level documentation. diff --git a/platform/extension/messagequeue/mysql/message_store.go b/platform/extension/messagequeue/mysql/message_store.go index 97bad2e9..3667b16c 100644 --- a/platform/extension/messagequeue/mysql/message_store.go +++ b/platform/extension/messagequeue/mysql/message_store.go @@ -44,15 +44,7 @@ func newMessageStore(db *sql.DB, logger *zap.SugaredLogger, scope tally.Scope) m } } -// Insert inserts messages into the messages table with no visibility delay. -// Equivalent to InsertDelayed with visibleAfterMs == 0. -func (s *sqlmessageStore) Insert(ctx context.Context, topic string, messages []entityqueue.Message) error { - return s.InsertDelayed(ctx, topic, messages, 0) -} - -// InsertDelayed inserts messages into the messages table, optionally deferring -// delivery until visibleAfterMs (epoch milliseconds). 0 means immediately -// visible; FetchByOffset skips rows where visible_after > now. +// Insert inserts messages into the messages table. // // Publishes are idempotent on the (topic, partition_key, id) unique key: a // repeated publish for the same key is silently treated as success and does @@ -61,7 +53,7 @@ func (s *sqlmessageStore) Insert(ctx context.Context, topic string, messages []e // idempotent publishes") and lets callers safely retry publishes (e.g. a // second Cancel RPC for the same request) without surfacing 1062 duplicate-key // errors. -func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messages []entityqueue.Message, visibleAfterMs int64) (retErr error) { +func (s *sqlmessageStore) Insert(ctx context.Context, topic string, messages []entityqueue.Message) (retErr error) { op := metrics.Begin(s.scope, "insert", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic)) defer func() { op.Complete(retErr) }() @@ -72,7 +64,6 @@ func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messa s.logger.Debugw("inserting messages", logTopic, topic, "count", len(messages), - "visible_after", visibleAfterMs, ) tx, err := s.db.BeginTx(ctx, nil) @@ -84,8 +75,8 @@ func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messa // ON DUPLICATE KEY UPDATE topic=topic is a no-op write that makes MySQL // swallow the unique-key violation without mutating the existing row. stmt, err := tx.PrepareContext(ctx, fmt.Sprintf(` - INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, visible_after, failed_at, failure_count, last_error, original_topic) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, 0, '', '') + INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, failed_at, failure_count, last_error, original_topic) + VALUES (?, ?, ?, ?, ?, ?, ?, 0, 0, '', '') ON DUPLICATE KEY UPDATE topic = topic `, MessagesTableName)) if err != nil { @@ -111,7 +102,6 @@ func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messa msg.PartitionKey, now, msg.PublishedAt, - visibleAfterMs, ) if err != nil { return fmt.Errorf("insert message topic=%s message=%s partition=%s: %w", topic, msg.ID, msg.PartitionKey, err) @@ -147,20 +137,18 @@ func (s *sqlmessageStore) Delete(ctx context.Context, topic string, partitionKey } // FetchByOffset fetches messages with offset > currentOffset for a specific partition. -// Rows whose visible_after > nowMs are skipped — those are deferred deliveries -// (published via InsertDelayed) that should not yet be surfaced to subscribers. // Messages are fetched from the immutable log; no per-message mutation occurs. -func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, partitionKey string, currentOffset int64, nowMs int64, limit int) (_ []messageRow, retErr error) { +func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, partitionKey string, currentOffset int64, limit int) (_ []messageRow, retErr error) { op := metrics.Begin(s.scope, "fetch", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic)) defer func() { op.Complete(retErr) }() rows, err := s.db.QueryContext(ctx, fmt.Sprintf(` SELECT offset, id, payload, metadata, partition_key, published_at, failed_at, failure_count, last_error, original_topic FROM %s - WHERE topic = ? AND partition_key = ? AND offset > ? AND visible_after <= ? + WHERE topic = ? AND partition_key = ? AND offset > ? ORDER BY offset LIMIT ? - `, MessagesTableName), topic, partitionKey, currentOffset, nowMs, limit) + `, MessagesTableName), topic, partitionKey, currentOffset, limit) if err != nil { return nil, fmt.Errorf("query messages topic=%s partition=%s: %w", topic, partitionKey, err) } @@ -267,12 +255,10 @@ func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partition } // Insert into queue_messages table with DLQ topic name and DLQ-specific fields. - // DLQ messages are always immediately visible (visible_after=0); any delay on - // the original message has already been consumed by the time it failed. now := time.Now().UnixMilli() _, err = tx.ExecContext(ctx, fmt.Sprintf(` - INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, visible_after, failed_at, failure_count, last_error, original_topic) - VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?) + INSERT INTO %s (topic, id, payload, metadata, partition_key, created_at, published_at, failed_at, failure_count, last_error, original_topic) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, MessagesTableName), dlqTopic, messageID, payload, metadataJSON, fetchPartKey, createdAtMilli, publishedAtMilli, now, failureCount, lastError, topic) if err != nil { diff --git a/platform/extension/messagequeue/mysql/message_store_test.go b/platform/extension/messagequeue/mysql/message_store_test.go index 7e8188cf..ca0913a1 100644 --- a/platform/extension/messagequeue/mysql/message_store_test.go +++ b/platform/extension/messagequeue/mysql/message_store_test.go @@ -141,7 +141,6 @@ func TestMessageStore_FetchByOffset(t *testing.T) { topic := "test_topic" partitionKey := "part1" currentOffset := int64(0) - nowMs := time.Now().UnixMilli() limit := 10 // Mock query results (no transaction, simple SELECT) @@ -149,60 +148,16 @@ func TestMessageStore_FetchByOffset(t *testing.T) { AddRow(int64(1), "msg1", []byte("payload1"), []byte("{}"), "part1", time.Now().UnixMilli(), int64(0), 0, "", "") mock.ExpectQuery("SELECT (.+) FROM queue_messages"). - WithArgs(topic, partitionKey, currentOffset, nowMs, limit). + WithArgs(topic, partitionKey, currentOffset, limit). WillReturnRows(rows) - results, err := store.FetchByOffset(ctx, topic, partitionKey, currentOffset, nowMs, limit) + results, err := store.FetchByOffset(ctx, topic, partitionKey, currentOffset, limit) require.NoError(t, err) require.Len(t, results, 1) require.Equal(t, "msg1", results[0].ID) require.NoError(t, mock.ExpectationsWereMet()) } -func TestMessageStore_FetchByOffset_SkipsDelayed(t *testing.T) { - db, mock, store := setupmessageStoreTest(t) - defer db.Close() - - ctx := context.Background() - topic := "test_topic" - partitionKey := "part1" - currentOffset := int64(0) - nowMs := int64(1000) - limit := 10 - - // The SQL filter (visible_after <= nowMs) is applied by the DB; sqlmock just - // verifies the parameter binding. An empty result row simulates the case - // where the only message is still deferred. - mock.ExpectQuery("SELECT (.+) FROM queue_messages"). - WithArgs(topic, partitionKey, currentOffset, nowMs, limit). - WillReturnRows(sqlmock.NewRows([]string{"offset", "id", "payload", "metadata", "partition_key", "published_at", "failed_at", "failure_count", "last_error", "original_topic"})) - - results, err := store.FetchByOffset(ctx, topic, partitionKey, currentOffset, nowMs, limit) - require.NoError(t, err) - require.Empty(t, results) - require.NoError(t, mock.ExpectationsWereMet()) -} - -func TestMessageStore_InsertDelayed(t *testing.T) { - db, mock, store := setupmessageStoreTest(t) - defer db.Close() - - ctx := context.Background() - visibleAfter := time.Now().UnixMilli() + 5000 - msg := entityqueue.Message{ID: "msg-delayed", Payload: []byte("p"), PartitionKey: "part1", PublishedAt: time.Now().UnixMilli()} - - mock.ExpectBegin() - mock.ExpectPrepare("INSERT INTO queue_messages") - mock.ExpectExec("INSERT INTO queue_messages"). - WithArgs("test_topic", msg.ID, msg.Payload, []byte(nil), msg.PartitionKey, sqlmock.AnyArg(), msg.PublishedAt, visibleAfter). - WillReturnResult(sqlmock.NewResult(1, 1)) - mock.ExpectCommit() - - err := store.InsertDelayed(ctx, "test_topic", []entityqueue.Message{msg}, visibleAfter) - require.NoError(t, err) - require.NoError(t, mock.ExpectationsWereMet()) -} - func TestMessageStore_MoveToDLQ(t *testing.T) { db, mock, store := setupmessageStoreTest(t) defer db.Close() diff --git a/platform/extension/messagequeue/mysql/mock_stores.go b/platform/extension/messagequeue/mysql/mock_stores.go index 87af009c..844c7af3 100644 --- a/platform/extension/messagequeue/mysql/mock_stores.go +++ b/platform/extension/messagequeue/mysql/mock_stores.go @@ -56,18 +56,18 @@ func (mr *MockmessageStoreMockRecorder) Delete(ctx, topic, partitionKey, message } // FetchByOffset mocks base method. -func (m *MockmessageStore) FetchByOffset(ctx context.Context, topic, partitionKey string, currentOffset, nowMs int64, limit int) ([]messageRow, error) { +func (m *MockmessageStore) FetchByOffset(ctx context.Context, topic, partitionKey string, currentOffset int64, limit int) ([]messageRow, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "FetchByOffset", ctx, topic, partitionKey, currentOffset, nowMs, limit) + ret := m.ctrl.Call(m, "FetchByOffset", ctx, topic, partitionKey, currentOffset, limit) ret0, _ := ret[0].([]messageRow) ret1, _ := ret[1].(error) return ret0, ret1 } // FetchByOffset indicates an expected call of FetchByOffset. -func (mr *MockmessageStoreMockRecorder) FetchByOffset(ctx, topic, partitionKey, currentOffset, nowMs, limit any) *gomock.Call { +func (mr *MockmessageStoreMockRecorder) FetchByOffset(ctx, topic, partitionKey, currentOffset, limit any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FetchByOffset", reflect.TypeOf((*MockmessageStore)(nil).FetchByOffset), ctx, topic, partitionKey, currentOffset, nowMs, limit) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FetchByOffset", reflect.TypeOf((*MockmessageStore)(nil).FetchByOffset), ctx, topic, partitionKey, currentOffset, limit) } // GarbageCollect mocks base method. @@ -114,20 +114,6 @@ func (mr *MockmessageStoreMockRecorder) Insert(ctx, topic, messages any) *gomock return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Insert", reflect.TypeOf((*MockmessageStore)(nil).Insert), ctx, topic, messages) } -// InsertDelayed mocks base method. -func (m *MockmessageStore) InsertDelayed(ctx context.Context, topic string, messages []messagequeue.Message, visibleAfterMs int64) error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "InsertDelayed", ctx, topic, messages, visibleAfterMs) - ret0, _ := ret[0].(error) - return ret0 -} - -// InsertDelayed indicates an expected call of InsertDelayed. -func (mr *MockmessageStoreMockRecorder) InsertDelayed(ctx, topic, messages, visibleAfterMs any) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InsertDelayed", reflect.TypeOf((*MockmessageStore)(nil).InsertDelayed), ctx, topic, messages, visibleAfterMs) -} - // MoveToDLQ mocks base method. func (m *MockmessageStore) MoveToDLQ(ctx context.Context, topic, partitionKey, messageID string, failureCount int, lastError, dlqTopicSuffix string) error { m.ctrl.T.Helper() diff --git a/platform/extension/messagequeue/mysql/publisher.go b/platform/extension/messagequeue/mysql/publisher.go index 225c8906..fc9ecfe3 100644 --- a/platform/extension/messagequeue/mysql/publisher.go +++ b/platform/extension/messagequeue/mysql/publisher.go @@ -18,7 +18,6 @@ import ( "context" "fmt" "sync" - "time" "github.com/uber-go/tally" "go.uber.org/zap" @@ -67,36 +66,6 @@ func (p *publisher) Publish(ctx context.Context, topic string, message entityque return nil } -// PublishAfter sends a message that becomes visible to subscribers only -// after delayMs from now. The message is inserted with visible_after = -// now + delayMs; FetchByOffset skips it until that timestamp. -// delayMs <= 0 is equivalent to Publish. -func (p *publisher) PublishAfter(ctx context.Context, topic string, message entityqueue.Message, delayMs int64) (retErr error) { - op := metrics.Begin(p.scope, "publish_after", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic)) - defer func() { op.Complete(retErr) }() - - p.mu.RLock() - closed := p.closed - p.mu.RUnlock() - - if closed { - return ErrPublisherClosed - } - - var visibleAfter int64 - if delayMs > 0 { - visibleAfter = time.Now().UnixMilli() + delayMs - } - - if err := p.messageStore.InsertDelayed(ctx, topic, []entityqueue.Message{message}, visibleAfter); err != nil { - return fmt.Errorf("publish_after message store insert error: %w", err) - } - - p.logger.Debugw("published delayed message", logTopic, topic, logMessageID, message.ID, "delay_ms", delayMs) - - return nil -} - // Close gracefully shuts down the publisher func (p *publisher) Close() error { p.mu.Lock() diff --git a/platform/extension/messagequeue/mysql/publisher_test.go b/platform/extension/messagequeue/mysql/publisher_test.go index f54af29d..115c7231 100644 --- a/platform/extension/messagequeue/mysql/publisher_test.go +++ b/platform/extension/messagequeue/mysql/publisher_test.go @@ -140,7 +140,7 @@ func TestPublisher_Publish(t *testing.T) { } } -func TestPublisher_PublishAfterClose(t *testing.T) { +func TestPublisher_PublishOnClosedPublisher(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -160,68 +160,6 @@ func TestPublisher_PublishAfterClose(t *testing.T) { require.True(t, errors.Is(err, ErrPublisherClosed)) } -func TestPublisher_PublishAfter(t *testing.T) { - tests := []struct { - name string - delayMs int64 - wantVisibleArg gomock.Matcher - }{ - { - name: "positive delay binds future visible_after", - delayMs: 5000, - // Exact timestamp depends on wall clock; assert it's > 0. - wantVisibleArg: gomock.Cond(func(v any) bool { - ts, ok := v.(int64) - return ok && ts > 0 - }), - }, - { - name: "zero delay binds visible_after=0", - delayMs: 0, - wantVisibleArg: gomock.Eq(int64(0)), - }, - { - name: "negative delay clamps to 0", - delayMs: -100, - wantVisibleArg: gomock.Eq(int64(0)), - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - ctrl := gomock.NewController(t) - defer ctrl.Finish() - - mockStore := NewMockmessageStore(ctrl) - mockStore.EXPECT(). - InsertDelayed(gomock.Any(), "test_topic", gomock.Any(), tt.wantVisibleArg). - Return(nil). - Times(1) - - pub := setupPublisherTest(t, mockStore) - - msg := entityqueue.NewMessage("msg-delayed", []byte("p"), "part1", nil) - err := pub.PublishAfter(context.Background(), "test_topic", msg, tt.delayMs) - require.NoError(t, err) - }) - } -} - -func TestPublisher_PublishAfterClosed(t *testing.T) { - ctrl := gomock.NewController(t) - defer ctrl.Finish() - - mockStore := NewMockmessageStore(ctrl) - pub := setupPublisherTest(t, mockStore) - - require.NoError(t, pub.Close()) - - msg := entityqueue.NewMessage("msg1", []byte("p"), "part1", nil) - err := pub.PublishAfter(context.Background(), "test_topic", msg, 1000) - require.Error(t, err) - require.True(t, errors.Is(err, ErrPublisherClosed)) -} - func TestPublisher_Close(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() diff --git a/platform/extension/messagequeue/mysql/schema/queue_messages.sql b/platform/extension/messagequeue/mysql/schema/queue_messages.sql index 8c2e7739..50e887b2 100644 --- a/platform/extension/messagequeue/mysql/schema/queue_messages.sql +++ b/platform/extension/messagequeue/mysql/schema/queue_messages.sql @@ -25,13 +25,6 @@ CREATE TABLE IF NOT EXISTS queue_messages ( created_at BIGINT UNSIGNED NOT NULL, published_at BIGINT UNSIGNED NOT NULL, - -- visible_after defers delivery: subscribers skip rows where visible_after > now. - -- 0 (the default) means immediately visible. Set by Publisher.PublishAfter - -- to schedule a fresh message for delivery at a future time without - -- consuming a delivery_state retry slot (used e.g. by the orchestrator's - -- buildstatus polling consumer to space out Status calls). - visible_after BIGINT UNSIGNED NOT NULL DEFAULT 0, - -- DLQ-specific fields (0/"" for normal messages, populated for DLQ messages) failed_at BIGINT UNSIGNED NOT NULL, -- failure_count stores how many times the message failed on the ORIGINAL topic before moving to DLQ diff --git a/platform/extension/messagequeue/mysql/stores.go b/platform/extension/messagequeue/mysql/stores.go index 424c5b8f..868da242 100644 --- a/platform/extension/messagequeue/mysql/stores.go +++ b/platform/extension/messagequeue/mysql/stores.go @@ -60,18 +60,12 @@ type messageStore interface { // Insert inserts messages into the topic table. Insert(ctx context.Context, topic string, messages []entityqueue.Message) error - // InsertDelayed inserts messages whose delivery is deferred until - // visibleAfterMs (epoch milliseconds). 0 means immediately visible - // and is equivalent to Insert. - InsertDelayed(ctx context.Context, topic string, messages []entityqueue.Message, visibleAfterMs int64) error - // Delete deletes a message by topic, partition key, and ID Delete(ctx context.Context, topic string, partitionKey string, messageID string) error // FetchByOffset fetches messages with offset > currentOffset for a specific partition. - // Rows whose visible_after > nowMs are skipped (deferred deliveries). // Per-consumer-group visibility is handled by the deliveryStateStore. - FetchByOffset(ctx context.Context, topic string, partitionKey string, currentOffset int64, nowMs int64, limit int) ([]messageRow, error) + FetchByOffset(ctx context.Context, topic string, partitionKey string, currentOffset int64, limit int) ([]messageRow, error) // MoveToDLQ moves a message to the dead letter queue // dlqTopicSuffix is appended to the original topic to form the DLQ topic name diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 62a5769c..5d5fd377 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -799,9 +799,8 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) { return fmt.Errorf("get acked offset: %w", err) } - // Fetch messages from immutable log; defer-visible rows (visible_after > now) - // are skipped at the SQL layer. - rows, err := s.messageStore.FetchByOffset(ctx, sub.topic, partitionKey, currentOffset, time.Now().UnixMilli(), cfg.BatchSize) + // Fetch messages from the immutable log. + rows, err := s.messageStore.FetchByOffset(ctx, sub.topic, partitionKey, currentOffset, cfg.BatchSize) if err != nil { return fmt.Errorf("fetch messages: %w", err) } diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index ea770783..d5c12cb5 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -484,7 +484,7 @@ func TestSubscriber_ReconcilePartitionWorkers(t *testing.T) { // Allow offset initialization, fetch, and watermark calls from workers mockOffsetStore.EXPECT().Initialize(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() mockOffsetStore.EXPECT().GetAckedOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes() - mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() + mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() mockMessageStore.EXPECT().GetOffsetsAbove(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() mockMessageStore.EXPECT().GarbageCollect(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes() mockOffsetStore.EXPECT().GetMinAckedOffset(gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), false, nil).AnyTimes() @@ -565,7 +565,7 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) { Payload: []byte("payload"), PublishedAt: time.Now().UnixMilli(), } - mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), "test_topic", "part-1", int64(0), gomock.Any(), cfg.BatchSize). + mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), "test_topic", "part-1", int64(0), cfg.BatchSize). Return([]messageRow{row}, nil) // Delivery state checks — GetDeliveryState returns not-found (new message) @@ -691,7 +691,7 @@ func TestSubscriber_PollAndDeliver_PostponedBarrier(t *testing.T) { {ID: "msg-2", Offset: 2, PartitionKey: "part-1", Payload: []byte("p2"), PublishedAt: time.Now().UnixMilli()}, {ID: "msg-3", Offset: 3, PartitionKey: "part-1", Payload: []byte("p3"), PublishedAt: time.Now().UnixMilli()}, } - mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), "test_topic", "part-1", int64(0), gomock.Any(), cfg.BatchSize). + mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), "test_topic", "part-1", int64(0), cfg.BatchSize). Return(rows, nil) mockDeliveryState.EXPECT().GetDeliveryState(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", int64(1)). @@ -753,7 +753,7 @@ func TestSubscriber_StopAllWorkers(t *testing.T) { // Allow worker polling and watermark advancement mockOffsetStore.EXPECT().Initialize(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() mockOffsetStore.EXPECT().GetAckedOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes() - mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() + mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() mockMessageStore.EXPECT().GetOffsetsAbove(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes() mockMessageStore.EXPECT().GarbageCollect(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes() mockOffsetStore.EXPECT().GetMinAckedOffset(gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), false, nil).AnyTimes() diff --git a/platform/extension/messagequeue/publisher.go b/platform/extension/messagequeue/publisher.go index 0a9f8034..6881189d 100644 --- a/platform/extension/messagequeue/publisher.go +++ b/platform/extension/messagequeue/publisher.go @@ -28,18 +28,6 @@ type Publisher interface { // Publish sends a message to the specified topic. Publish(ctx context.Context, topic string, message entityqueue.Message) error - // PublishAfter sends a message that becomes visible to subscribers only - // after delayMs from now. It is a fresh publish — not a redelivery — so - // it does not consume a delivery_state retry slot. delayMs <= 0 is - // equivalent to Publish. - // - // Use for deferring *other* work — a delayed message to another topic or - // key. A consumer deferring its own current delivery should use - // Delivery.Postpone instead; use Nack with a delay for "this delivery - // failed, try again" — the signals stay separate so retry_count and DLQ - // behaviour remain meaningful. - PublishAfter(ctx context.Context, topic string, message entityqueue.Message, delayMs int64) error - // Close gracefully shuts down the publisher, flushing pending messages. Close() error }