Read pending scheduled posts from master to avoid replica lag gaps (#37104)

* Fix scheduled post job to fetch pending posts from master

The scheduled post job fetches a page of pending posts, deletes the
processed ones, then fetches the next page. Reading the page from a
read replica can return stale data, causing already-processed posts to
reappear on later pages. Read from master to avoid this gap.

Co-authored-by: mattermost-code <matty-code@mattermost.com>

* Add regression test for scheduled post master read

Verifies GetPendingScheduledPosts reads pending posts from the master
database by pointing the replica at a separate empty database; a read
from the replica would return no rows.

Co-authored-by: mattermost-code <matty-code@mattermost.com>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: mattermost-code <matty-code@mattermost.com>

server/channels/store/sqlstore/scheduled_post_store.go

@@ -179,8 +179,11 @@ func (s *SqlScheduledPostStore) GetPendingScheduledPosts(beforeTime, afterTime i
 			})
 	}
 
+	// We read from the master here instead of a replica on purpose. The scheduled post job
+	// deletes processed posts and then fetches the next page of pending posts. Reading from a
+	// replica can return stale data, causing already-processed posts to reappear on later pages.
 	var scheduledPosts []*model.ScheduledPost
-	if err := s.GetReplica().SelectBuilder(&scheduledPosts, query); err != nil {
+	if err := s.GetMaster().SelectBuilder(&scheduledPosts, query); err != nil {
 		mlog.Error(
 			"SqlScheduledPostStore.GetPendingScheduledPosts: failed to fetch pending scheduled posts for processing",
 			mlog.Int("before_time", beforeTime),

server/channels/store/sqlstore/scheduled_post_store_test.go

@@ -6,9 +6,80 @@ package sqlstore
 import (
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
 	"github.com/mattermost/mattermost/server/v8/channels/store/storetest"
 )
 
 func TestScheduledPostStore(t *testing.T) {
 	StoreTestWithSqlStore(t, storetest.TestScheduledPostStore)
 }
+
+// TestGetPendingScheduledPostsReadsFromMaster verifies that the scheduled post job reads its
+// pending posts from the master database rather than a read replica. The job deletes processed
+// posts and then fetches the next page, so reading from a lagging replica could surface
+// already-processed posts again. The test points the replica at a separate, empty database so
+// that a read from the replica would return nothing, proving the read targets master.
+func TestGetPendingScheduledPostsReadsFromMaster(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping test in short mode")
+	}
+
+	logger := mlog.CreateTestLogger(t)
+
+	masterSettings, err := makeSqlSettings(model.DatabaseDriverPostgres)
+	if err != nil {
+		t.Skip(err)
+	}
+	defer storetest.CleanupSqlSettings(masterSettings)
+
+	// The replica lives in its own database. We migrate it so the ScheduledPosts table exists
+	// but leave it empty: the row under test will only ever exist on master.
+	replicaSettings, err := makeSqlSettings(model.DatabaseDriverPostgres)
+	if err != nil {
+		t.Skip(err)
+	}
+	defer storetest.CleanupSqlSettings(replicaSettings)
+
+	replicaStore, err := New(*replicaSettings, logger, nil)
+	require.NoError(t, err)
+	replicaStore.Close()
+
+	masterSettings.DataSourceReplicas = []string{*replicaSettings.DataSource}
+
+	store, err := New(*masterSettings, logger, nil)
+	require.NoError(t, err)
+	defer store.Close()
+
+	// A license is required for replica reads to route to the replica. Without one, GetReplica()
+	// falls back to master and the test could pass even if the code read from the replica.
+	store.UpdateLicense(&model.License{})
+	require.NotSame(t, store.GetMaster(), store.GetReplica(), "replica must be a distinct connection for this test to be meaningful")
+
+	scheduledPost := &model.ScheduledPost{
+		Draft: model.Draft{
+			CreateAt:  model.GetMillis(),
+			UserId:    model.NewId(),
+			ChannelId: model.NewId(),
+			Message:   "pending scheduled post",
+		},
+		ScheduledAt: model.GetMillis(),
+	}
+	createdScheduledPost, err := store.ScheduledPost().CreateScheduledPost(scheduledPost)
+	require.NoError(t, err)
+	require.NotEmpty(t, createdScheduledPost.Id)
+
+	// Sanity check: the replica database does not contain the scheduled post.
+	var replicaCount int
+	require.NoError(t, store.GetReplica().Get(&replicaCount, "SELECT COUNT(*) FROM ScheduledPosts"))
+	require.Zero(t, replicaCount, "replica should be empty so the test can distinguish master from replica reads")
+
+	beforeTime := createdScheduledPost.ScheduledAt + 1000
+	afterTime := createdScheduledPost.ScheduledAt - (24 * 60 * 60 * 1000)
+	pending, err := store.ScheduledPost().GetPendingScheduledPosts(beforeTime, afterTime, "", 10)
+	require.NoError(t, err)
+	require.Len(t, pending, 1, "pending posts must be read from master, not the empty replica")
+	require.Equal(t, createdScheduledPost.Id, pending[0].Id)
+}