feat(messagequeue)!: remove PublishAfter and the visible_after column

## Summary

### Why?

PublishAfter existed for one pattern: a consumer acking its own delivery and republishing a delayed copy to keep a poll or gate-wait loop alive. The two previous commits migrated all three such loops (stovepipe process/buildsignal, orchestrator buildsignal) to the hold primitive, leaving PublishAfter with zero callers. Removing it shrinks the publisher contract to a single verb and deletes the delayed-visibility machinery that existed only to serve it.

### What?

Publisher loses PublishAfter; the mysql impl deletes its method, folds InsertDelayed back into Insert, drops FetchByOffset's nowMs parameter and its visible_after filter, removes the column from the queue_messages schema and the MoveToDLQ column list, and regenerates mocks. Tests covering delayed publish/fetch are deleted; the misnamed TestPublisher_PublishAfterClose (which tests Publish on a closed publisher) is renamed. READMEs drop the PublishAfter guidance — a consumer defers its own delivery with Hold/Postpone, and no caller needed cross-topic delayed publishing.

Deployment note: the schema files here are applied by dev/test stacks; a live database needs `ALTER TABLE queue_messages DROP COLUMN visible_after` (old rows with a pending visible_after would become immediately visible on rollout, which is the correct degradation — those rows were minted by the pre-hold republish pattern and their consumers supersede/no-op them).

## Test Plan

✅ `make test` (83 targets). ✅ `bazel test //test/integration/extension/messagequeue/...` — full queue behavior including idempotent publish, DLQ, postpone barrier. ✅ `make fmt`, `make mocks`.

Author: behinddwalls

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.
 

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.
 

platform/extension/messagequeue/mock/publisher_mock.go

@@ -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.

platform/extension/messagequeue/mysql/README.md

@@ -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 {

platform/extension/messagequeue/mysql/message_store.go

@@ -141,68 +141,23 @@ 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)
 	rows := sqlmock.NewRows([]string{"offset", "id", "payload", "metadata", "partition_key", "published_at", "failed_at", "failure_count", "last_error", "original_topic"}).
 		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()

platform/extension/messagequeue/mysql/message_store_test.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()

platform/extension/messagequeue/mysql/mock_stores.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()

platform/extension/messagequeue/mysql/publisher.go

@@ -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