[MM-53968] Includes mattermost-plugin-api into the mono repo (#24235)

Include https://github.com/mattermost/mattermost-plugin-api into the mono repo

Co-authored-by: Jesse Hallam <jesse.hallam@gmail.com>
Co-authored-by: Michael Kochell <mjkochell@gmail.com>
Co-authored-by: Alejandro García Montoro <alejandro.garciamontoro@gmail.com>
Co-authored-by: Ben Schumacher <ben.schumacher@mattermost.com>
Co-authored-by: Alex Dovenmuehle <alex.dovenmuehle@mattermost.com>
Co-authored-by: Michael Kochell <6913320+mickmister@users.noreply.github.com>
Co-authored-by: Christopher Poile <cpoile@gmail.com>
Co-authored-by: İlker Göktuğ Öztürk <ilkergoktugozturk@gmail.com>
Co-authored-by: Shota Gvinepadze <wineson@gmail.com>
Co-authored-by: Ali Farooq <ali.farooq0@pm.me>
Co-authored-by: Maria A Nunez <maria.nunez@mattermost.com>
Co-authored-by: Daniel Espino García <larkox@gmail.com>
Co-authored-by: Christopher Speller <crspeller@gmail.com>
Co-authored-by: Alex Dovenmuehle <adovenmuehle@gmail.com>
Co-authored-by: Szymon Gibała <szymongib@gmail.com>
Co-authored-by: Lev <1187448+levb@users.noreply.github.com>
Co-authored-by: Jason Frerich <jason.frerich@mattermost.com>
Co-authored-by: Agniva De Sarker <agnivade@yahoo.co.in>
Co-authored-by: Artur M. Wolff <artur.m.wolff@gmail.com>
Co-authored-by: Madhav Hugar <16546715+madhavhugar@users.noreply.github.com>
Co-authored-by: Joe <security.joe@pm.me>
Co-authored-by: Ibrahim Serdar Acikgoz <serdaracikgoz86@gmail.com>
Co-authored-by: José Peso <trilopin@users.noreply.github.com>

server/public/pluginapi/cluster/doc.go

@@ -0,0 +1,3 @@
+// package cluster exposes synchronization primitives to ensure correct behavior across multiple
+// plugin instances in a Mattermost cluster.
+package cluster

server/public/pluginapi/cluster/job.go

@@ -0,0 +1,229 @@
+package cluster
+
+import (
+	"encoding/json"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+const (
+	// cronPrefix is used to namespace key values created for a job from other key values
+	// created by a plugin.
+	cronPrefix = "cron_"
+)
+
+// JobPluginAPI is the plugin API interface required to schedule jobs.
+type JobPluginAPI interface {
+	MutexPluginAPI
+	KVGet(key string) ([]byte, *model.AppError)
+	KVDelete(key string) *model.AppError
+	KVList(page, count int) ([]string, *model.AppError)
+}
+
+// JobConfig defines the configuration of a scheduled job.
+type JobConfig struct {
+	// Interval is the period of execution for the job.
+	Interval time.Duration
+}
+
+// NextWaitInterval is a callback computing the next wait interval for a job.
+type NextWaitInterval func(now time.Time, metadata JobMetadata) time.Duration
+
+// MakeWaitForInterval creates a function to scheduling a job to run on the given interval relative
+// to the last finished timestamp.
+//
+// For example, if the job first starts at 12:01 PM, and is configured with interval 5 minutes,
+// it will next run at:
+//
+//	12:06, 12:11, 12:16, ...
+//
+// If the job has not previously started, it will run immediately.
+func MakeWaitForInterval(interval time.Duration) NextWaitInterval {
+	if interval == 0 {
+		panic("must specify non-zero ready interval")
+	}
+
+	return func(now time.Time, metadata JobMetadata) time.Duration {
+		sinceLastFinished := now.Sub(metadata.LastFinished)
+		if sinceLastFinished < interval {
+			return interval - sinceLastFinished
+		}
+
+		return 0
+	}
+}
+
+// MakeWaitForRoundedInterval creates a function, scheduling a job to run on the nearest rounded
+// interval relative to the last finished timestamp.
+//
+// For example, if the job first starts at 12:04 PM, and is configured with interval 5 minutes,
+// and is configured to round to 5 minute intervals, it will next run at:
+//
+//	12:05 PM, 12:10 PM, 12:15 PM, ...
+//
+// If the job has not previously started, it will run immediately. Note that this wait interval
+// strategy does not guarantee a minimum interval between runs, only that subsequent runs will be
+// scheduled on the rounded interval.
+func MakeWaitForRoundedInterval(interval time.Duration) NextWaitInterval {
+	if interval == 0 {
+		panic("must specify non-zero ready interval")
+	}
+
+	return func(now time.Time, metadata JobMetadata) time.Duration {
+		if metadata.LastFinished.IsZero() {
+			return 0
+		}
+
+		target := metadata.LastFinished.Add(interval).Truncate(interval)
+		untilTarget := target.Sub(now)
+		if untilTarget > 0 {
+			return untilTarget
+		}
+
+		return 0
+	}
+}
+
+// Job is a scheduled job whose callback function is executed on a configured interval by at most
+// one plugin instance at a time.
+//
+// Use scheduled jobs to perform background activity on a regular interval without having to
+// explicitly coordinate with other instances of the same plugin that might repeat that effort.
+type Job struct {
+	pluginAPI        JobPluginAPI
+	key              string
+	mutex            *Mutex
+	nextWaitInterval NextWaitInterval
+	callback         func()
+
+	stopOnce sync.Once
+	stop     chan bool
+	done     chan bool
+}
+
+// JobMetadata persists metadata about job execution.
+type JobMetadata struct {
+	// LastFinished is the last time the job finished anywhere in the cluster.
+	LastFinished time.Time
+}
+
+// Schedule creates a scheduled job.
+func Schedule(pluginAPI JobPluginAPI, key string, nextWaitInterval NextWaitInterval, callback func()) (*Job, error) {
+	key = cronPrefix + key
+
+	mutex, err := NewMutex(pluginAPI, key)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create job mutex")
+	}
+
+	job := &Job{
+		pluginAPI:        pluginAPI,
+		key:              key,
+		mutex:            mutex,
+		nextWaitInterval: nextWaitInterval,
+		callback:         callback,
+		stop:             make(chan bool),
+		done:             make(chan bool),
+	}
+
+	go job.run()
+
+	return job, nil
+}
+
+// readMetadata reads the job execution metadata from the kv store.
+func (j *Job) readMetadata() (JobMetadata, error) {
+	data, appErr := j.pluginAPI.KVGet(j.key)
+	if appErr != nil {
+		return JobMetadata{}, errors.Wrap(appErr, "failed to read data")
+	}
+
+	if data == nil {
+		return JobMetadata{}, nil
+	}
+
+	var metadata JobMetadata
+	err := json.Unmarshal(data, &metadata)
+	if err != nil {
+		return JobMetadata{}, errors.Wrap(err, "failed to decode data")
+	}
+
+	return metadata, nil
+}
+
+// saveMetadata writes updated job execution metadata from the kv store.
+//
+// It is assumed that the job mutex is held, negating the need to require an atomic write.
+func (j *Job) saveMetadata(metadata JobMetadata) error {
+	data, err := json.Marshal(metadata)
+	if err != nil {
+		return errors.Wrap(err, "failed to marshal data")
+	}
+
+	ok, appErr := j.pluginAPI.KVSetWithOptions(j.key, data, model.PluginKVSetOptions{})
+	if appErr != nil || !ok {
+		return errors.Wrap(appErr, "failed to set data")
+	}
+
+	return nil
+}
+
+// run attempts to run the scheduled job, guaranteeing only one instance is executing concurrently.
+func (j *Job) run() {
+	defer close(j.done)
+
+	var waitInterval time.Duration
+
+	for {
+		select {
+		case <-j.stop:
+			return
+		case <-time.After(waitInterval):
+		}
+
+		func() {
+			// Acquire the corresponding job lock and hold it throughout execution.
+			j.mutex.Lock()
+			defer j.mutex.Unlock()
+
+			metadata, err := j.readMetadata()
+			if err != nil {
+				j.pluginAPI.LogError("failed to read job metadata", "err", err, "key", j.key)
+				waitInterval = nextWaitInterval(waitInterval, err)
+				return
+			}
+
+			// Is it time to run the job?
+			waitInterval = j.nextWaitInterval(time.Now(), metadata)
+			if waitInterval > 0 {
+				return
+			}
+
+			// Run the job
+			j.callback()
+
+			metadata.LastFinished = time.Now()
+
+			err = j.saveMetadata(metadata)
+			if err != nil {
+				j.pluginAPI.LogError("failed to write job data", "err", err, "key", j.key)
+			}
+
+			waitInterval = j.nextWaitInterval(time.Now(), metadata)
+		}()
+	}
+}
+
+// Close terminates a scheduled job, preventing it from being scheduled on this plugin instance.
+func (j *Job) Close() error {
+	j.stopOnce.Do(func() {
+		close(j.stop)
+	})
+	<-j.done
+
+	return nil
+}

server/public/pluginapi/cluster/job_example_test.go

@@ -0,0 +1,25 @@
+package cluster
+
+import (
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/plugin"
+)
+
+func ExampleSchedule() {
+	// Use p.API from your plugin instead.
+	pluginAPI := plugin.API(nil)
+
+	callback := func() {
+		// periodic work to do
+	}
+
+	job, err := Schedule(pluginAPI, "key", MakeWaitForInterval(5*time.Minute), callback)
+	if err != nil {
+		panic("failed to schedule job")
+	}
+
+	// main thread
+
+	defer job.Close()
+}

server/public/pluginapi/cluster/job_once.go

@@ -0,0 +1,235 @@
+package cluster
+
+import (
+	"encoding/json"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+const (
+	// oncePrefix is used to namespace key values created for a scheduleOnce job
+	oncePrefix = "once_"
+
+	// keysPerPage is the maximum number of keys to retrieve from the db per call
+	keysPerPage = 1000
+
+	// maxNumFails is the maximum number of KVStore read fails or failed attempts to run the
+	// callback until the scheduler cancels a job.
+	maxNumFails = 3
+
+	// waitAfterFail is the amount of time to wait after a failure
+	waitAfterFail = 1 * time.Second
+
+	// pollNewJobsInterval is the amount of time to wait between polling the db for new scheduled jobs
+	pollNewJobsInterval = 5 * time.Minute
+
+	// scheduleOnceJitter is the range of jitter to add to intervals to avoid contention issues
+	scheduleOnceJitter = 100 * time.Millisecond
+
+	// propsLimit is the maximum length in bytes of the json-representation of a job's props.
+	// It exists to prevent job go rountines from consuming too much memory, as they are long running.
+	propsLimit = 10000
+)
+
+type JobOnceMetadata struct {
+	Key   string
+	RunAt time.Time
+	Props any
+}
+
+type JobOnce struct {
+	pluginAPI    JobPluginAPI
+	clusterMutex *Mutex
+
+	// key is the original key. It is prefixed with oncePrefix when used as a key in the KVStore
+	key      string
+	props    any
+	runAt    time.Time
+	numFails int
+
+	// done signals the job.run go routine to exit
+	done     chan bool
+	doneOnce sync.Once
+
+	// join is a join point for the job.run() goroutine to join the calling goroutine (in this case,
+	// the one calling job.Cancel)
+	join     chan bool
+	joinOnce sync.Once
+
+	storedCallback *syncedCallback
+	activeJobs     *syncedJobs
+}
+
+// Cancel terminates a scheduled job, preventing it from being scheduled on this plugin instance.
+// It also removes the job from the db, preventing it from being run in the future.
+func (j *JobOnce) Cancel() {
+	j.clusterMutex.Lock()
+	defer j.clusterMutex.Unlock()
+
+	j.cancelWhileHoldingMutex()
+
+	// join the running goroutine
+	j.joinOnce.Do(func() {
+		<-j.join
+	})
+}
+
+func newJobOnce(pluginAPI JobPluginAPI, key string, runAt time.Time, callback *syncedCallback, jobs *syncedJobs, props any) (*JobOnce, error) {
+	mutex, err := NewMutex(pluginAPI, key)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create job mutex")
+	}
+
+	propsBytes, err := json.Marshal(props)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to marshal props")
+	}
+
+	if len(propsBytes) > propsLimit {
+		return nil, errors.Errorf("props length extends limit")
+	}
+
+	return &JobOnce{
+		pluginAPI:      pluginAPI,
+		clusterMutex:   mutex,
+		key:            key,
+		props:          props,
+		runAt:          runAt,
+		done:           make(chan bool),
+		join:           make(chan bool),
+		storedCallback: callback,
+		activeJobs:     jobs,
+	}, nil
+}
+
+func (j *JobOnce) run() {
+	defer close(j.join)
+
+	wait := time.Until(j.runAt)
+
+	for {
+		select {
+		case <-j.done:
+			return
+		case <-time.After(wait + addJitter()):
+		}
+
+		func() {
+			// Acquire the cluster mutex while we're trying to do the job
+			j.clusterMutex.Lock()
+			defer j.clusterMutex.Unlock()
+
+			// Check that the job has not been completed
+			metadata, err := readMetadata(j.pluginAPI, j.key)
+			if err != nil {
+				j.numFails++
+				if j.numFails > maxNumFails {
+					j.cancelWhileHoldingMutex()
+					return
+				}
+
+				// wait a bit of time and try again
+				wait = waitAfterFail
+				return
+			}
+
+			// If key doesn't exist, or if the runAt has changed, the original job has been completed already
+			if metadata == nil || !j.runAt.Equal(metadata.RunAt) {
+				j.cancelWhileHoldingMutex()
+				return
+			}
+
+			j.executeJob()
+
+			j.cancelWhileHoldingMutex()
+		}()
+	}
+}
+
+func (j *JobOnce) executeJob() {
+	j.storedCallback.mu.Lock()
+	defer j.storedCallback.mu.Unlock()
+
+	j.storedCallback.callback(j.key, j.props)
+}
+
+// readMetadata reads the job's stored metadata. If the caller wishes to make an atomic
+// read/write, the cluster mutex for job's key should be held.
+func readMetadata(pluginAPI JobPluginAPI, key string) (*JobOnceMetadata, error) {
+	data, appErr := pluginAPI.KVGet(oncePrefix + key)
+	if appErr != nil {
+		return nil, errors.Wrap(normalizeAppErr(appErr), "failed to read data")
+	}
+
+	if data == nil {
+		return nil, nil
+	}
+
+	var metadata JobOnceMetadata
+	if err := json.Unmarshal(data, &metadata); err != nil {
+		return nil, errors.Wrap(err, "failed to decode data")
+	}
+
+	return &metadata, nil
+}
+
+// saveMetadata writes the job's metadata to the kvstore. saveMetadata acquires the job's cluster lock.
+// saveMetadata will not overwrite an existing key.
+func (j *JobOnce) saveMetadata() error {
+	j.clusterMutex.Lock()
+	defer j.clusterMutex.Unlock()
+
+	metadata := JobOnceMetadata{
+		Key:   j.key,
+		Props: j.props,
+		RunAt: j.runAt,
+	}
+	data, err := json.Marshal(metadata)
+	if err != nil {
+		return errors.Wrap(err, "failed to marshal data")
+	}
+
+	ok, appErr := j.pluginAPI.KVSetWithOptions(oncePrefix+j.key, data, model.PluginKVSetOptions{
+		Atomic:   true,
+		OldValue: nil,
+	})
+	if appErr != nil {
+		return normalizeAppErr(appErr)
+	}
+	if !ok {
+		return errors.New("failed to set data")
+	}
+
+	return nil
+}
+
+// cancelWhileHoldingMutex assumes the caller holds the job's mutex.
+func (j *JobOnce) cancelWhileHoldingMutex() {
+	// remove the job from the kv store, if it exists
+	_ = j.pluginAPI.KVDelete(oncePrefix + j.key)
+
+	j.activeJobs.mu.Lock()
+	defer j.activeJobs.mu.Unlock()
+	delete(j.activeJobs.jobs, j.key)
+
+	j.doneOnce.Do(func() {
+		close(j.done)
+	})
+}
+
+func addJitter() time.Duration {
+	return time.Duration(rand.Int63n(int64(scheduleOnceJitter)))
+}
+
+func normalizeAppErr(appErr *model.AppError) error {
+	if appErr == nil {
+		return nil
+	}
+
+	return appErr
+}

server/public/pluginapi/cluster/job_once_example_test.go

@@ -0,0 +1,45 @@
+package cluster
+
+import (
+	"log"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/plugin"
+)
+
+func HandleJobOnceCalls(key string, props any) {
+	if key == "the key i'm watching for" {
+		log.Println(props)
+		// Work to do only once per cluster
+	}
+}
+
+func ExampleJobOnceScheduler_ScheduleOnce() {
+	// Use p.API from your plugin instead.
+	pluginAPI := plugin.API(nil)
+
+	// Get the scheduler, which you can pass throughout the plugin...
+	scheduler := GetJobOnceScheduler(pluginAPI)
+
+	// Set the plugin's callback handler
+	_ = scheduler.SetCallback(HandleJobOnceCalls)
+
+	// Now start the scheduler, which starts the poller and schedules all waiting jobs.
+	_ = scheduler.Start()
+
+	// main thread...
+
+	// add a job
+	_, _ = scheduler.ScheduleOnce("the key i'm watching for", time.Now().Add(2*time.Hour), struct{ foo string }{"aasd"})
+
+	// Maybe you want to check the scheduled jobs, or cancel them. This is completely optional--there
+	// is no need to cancel jobs, even if you are shutting down. Call Cancel only when you want to
+	// cancel a future job. Cancelling a job will prevent it from running in the future on this or
+	// any server.
+	jobs, _ := scheduler.ListScheduledJobs()
+	defer func() {
+		for _, j := range jobs {
+			scheduler.Cancel(j.Key)
+		}
+	}()
+}

server/public/pluginapi/cluster/job_once_mem_test.go

@@ -0,0 +1,83 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package cluster
+
+import (
+	"fmt"
+	"runtime"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+func TestMemFootprint(t *testing.T) {
+	var memConsumed = func() uint64 {
+		runtime.GC()
+		var s runtime.MemStats
+		runtime.ReadMemStats(&s)
+		return s.Sys
+	}
+
+	t.Run("average k per jobOnce", func(t *testing.T) {
+		t.SkipNow()
+
+		makeKey := model.NewId
+
+		numJobs := 100000
+		jobs := make(map[string]*int32, numJobs)
+		for i := 0; i < numJobs; i++ {
+			jobs[makeKey()] = new(int32)
+		}
+
+		callback := func(key string, _ any) {
+			count, ok := jobs[key]
+			if ok {
+				atomic.AddInt32(count, 1)
+			}
+		}
+
+		mockPluginAPI := newMockPluginAPI(t)
+		s := GetJobOnceScheduler(mockPluginAPI)
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		err = s.Start()
+		require.NoError(t, err)
+
+		getVal := func(key string) []byte {
+			data, _ := s.pluginAPI.KVGet(key)
+			return data
+		}
+
+		before := memConsumed()
+
+		for k := range jobs {
+			assert.Empty(t, getVal(oncePrefix+k))
+			_, err = s.ScheduleOnce(k, time.Now().Add(5*time.Minute), nil)
+			require.NoError(t, err)
+			assert.NotEmpty(t, getVal(oncePrefix+k))
+		}
+
+		time.Sleep(10 * time.Second)
+
+		// Everything scheduled now:
+		s.activeJobs.mu.RLock()
+		assert.Equal(t, numJobs, len(s.activeJobs.jobs))
+		s.activeJobs.mu.RUnlock()
+		list, err := s.ListScheduledJobs()
+		require.NoError(t, err)
+		assert.Equal(t, numJobs, len(list))
+
+		after := memConsumed()
+
+		fmt.Printf("\nthe %d jobs, scheduler, and goroutines require: %.2fmB memory, or %.3fkB each job\n",
+			numJobs,
+			float64(after-before)/(1024*1024),
+			(float64(after-before)/float64(numJobs))/1024)
+	})
+}

server/public/pluginapi/cluster/job_once_scheduler.go

@@ -0,0 +1,236 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package cluster
+
+import (
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+)
+
+// syncedCallback uses the mutex to make things predictable for the client: the callback will be
+// called once at a time (the client does not need to worry about concurrency within the callback)
+type syncedCallback struct {
+	mu       sync.Mutex
+	callback func(string, any)
+}
+
+type syncedJobs struct {
+	mu   sync.RWMutex
+	jobs map[string]*JobOnce
+}
+
+type JobOnceScheduler struct {
+	pluginAPI JobPluginAPI
+
+	startedMu sync.RWMutex
+	started   bool
+
+	activeJobs     *syncedJobs
+	storedCallback *syncedCallback
+}
+
+var schedulerOnce sync.Once
+var s *JobOnceScheduler
+
+// GetJobOnceScheduler returns a scheduler which is ready to have its callback set. Repeated
+// calls will return the same scheduler.
+func GetJobOnceScheduler(pluginAPI JobPluginAPI) *JobOnceScheduler {
+	schedulerOnce.Do(func() {
+		s = &JobOnceScheduler{
+			pluginAPI: pluginAPI,
+			activeJobs: &syncedJobs{
+				jobs: make(map[string]*JobOnce),
+			},
+			storedCallback: &syncedCallback{},
+		}
+	})
+	return s
+}
+
+// Start starts the Scheduler. It finds all previous ScheduleOnce jobs and starts them running, and
+// fires any jobs that have reached or exceeded their runAt time. Thus, even if a cluster goes down
+// and is restarted, Start will restart previously scheduled jobs.
+func (s *JobOnceScheduler) Start() error {
+	s.startedMu.Lock()
+	defer s.startedMu.Unlock()
+	if s.started {
+		return errors.New("scheduler has already been started")
+	}
+
+	if err := s.verifyCallbackExists(); err != nil {
+		return errors.Wrap(err, "callback not found; cannot start scheduler")
+	}
+
+	if err := s.scheduleNewJobsFromDB(); err != nil {
+		return errors.Wrap(err, "could not start JobOnceScheduler due to error")
+	}
+
+	go s.pollForNewScheduledJobs()
+
+	s.started = true
+
+	return nil
+}
+
+// SetCallback sets the scheduler's callback. When a job fires, the callback will be called with
+// the job's id.
+func (s *JobOnceScheduler) SetCallback(callback func(string, any)) error {
+	if callback == nil {
+		return errors.New("callback cannot be nil")
+	}
+
+	s.storedCallback.mu.Lock()
+	defer s.storedCallback.mu.Unlock()
+
+	s.storedCallback.callback = callback
+	return nil
+}
+
+// ListScheduledJobs returns a list of the jobs in the db that have been scheduled. There is no
+// guarantee that list is accurate by the time the caller reads the list. E.g., the jobs in the list
+// may have been run, canceled, or new jobs may have scheduled.
+func (s *JobOnceScheduler) ListScheduledJobs() ([]JobOnceMetadata, error) {
+	var ret []JobOnceMetadata
+	for i := 0; ; i++ {
+		keys, err := s.pluginAPI.KVList(i, keysPerPage)
+		if err != nil {
+			return nil, errors.Wrap(err, "error getting KVList")
+		}
+		for _, k := range keys {
+			if strings.HasPrefix(k, oncePrefix) {
+				metadata, err := readMetadata(s.pluginAPI, k[len(oncePrefix):])
+				if err != nil {
+					s.pluginAPI.LogError(errors.Wrap(err, "could not retrieve data from plugin kvstore for key: "+k).Error())
+					continue
+				}
+				if metadata == nil {
+					continue
+				}
+
+				ret = append(ret, *metadata)
+			}
+		}
+
+		if len(keys) < keysPerPage {
+			break
+		}
+	}
+
+	return ret, nil
+}
+
+// ScheduleOnce creates a scheduled job that will run once. When the clock reaches runAt, the
+// callback will be called with key and props as the argument.
+//
+// If the job key already exists in the db, this will return an error. To reschedule a job, first
+// cancel the original then schedule it again.
+func (s *JobOnceScheduler) ScheduleOnce(key string, runAt time.Time, props any) (*JobOnce, error) {
+	s.startedMu.RLock()
+	defer s.startedMu.RUnlock()
+	if !s.started {
+		return nil, errors.New("start the scheduler before adding jobs")
+	}
+
+	job, err := newJobOnce(s.pluginAPI, key, runAt, s.storedCallback, s.activeJobs, props)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not create new job")
+	}
+
+	if err = job.saveMetadata(); err != nil {
+		return nil, errors.Wrap(err, "could not save job metadata")
+	}
+
+	s.runAndTrack(job)
+
+	return job, nil
+}
+
+// Cancel cancels a job by its key. This is useful if the plugin lost the original *JobOnce, or
+// is stopping a job found in ListScheduledJobs().
+func (s *JobOnceScheduler) Cancel(key string) {
+	// using an anonymous function because job.Close() below needs access to the activeJobs mutex
+	job := func() *JobOnce {
+		s.activeJobs.mu.RLock()
+		defer s.activeJobs.mu.RUnlock()
+		j, ok := s.activeJobs.jobs[key]
+		if ok {
+			return j
+		}
+
+		// Job wasn't active, so no need to call CancelWhileHoldingMutex (which shuts down the
+		// goroutine). There's a condition where another server in the cluster started the job, and
+		// the current server hasn't polled for it yet. To solve that case, delete it from the db.
+		mutex, err := NewMutex(s.pluginAPI, key)
+		if err != nil {
+			s.pluginAPI.LogError(errors.Wrap(err, "failed to create job mutex in Cancel for key: "+key).Error())
+		}
+		mutex.Lock()
+		defer mutex.Unlock()
+
+		_ = s.pluginAPI.KVDelete(oncePrefix + key)
+
+		return nil
+	}()
+
+	if job != nil {
+		job.Cancel()
+	}
+}
+
+func (s *JobOnceScheduler) scheduleNewJobsFromDB() error {
+	scheduled, err := s.ListScheduledJobs()
+	if err != nil {
+		return errors.Wrap(err, "could not read scheduled jobs from db")
+	}
+
+	for _, m := range scheduled {
+		job, err := newJobOnce(s.pluginAPI, m.Key, m.RunAt, s.storedCallback, s.activeJobs, m.Props)
+		if err != nil {
+			s.pluginAPI.LogError(errors.Wrap(err, "could not create new job for key: "+m.Key).Error())
+			continue
+		}
+
+		s.runAndTrack(job)
+	}
+
+	return nil
+}
+
+func (s *JobOnceScheduler) runAndTrack(job *JobOnce) {
+	s.activeJobs.mu.Lock()
+	defer s.activeJobs.mu.Unlock()
+
+	// has this been scheduled already on this server?
+	if _, ok := s.activeJobs.jobs[job.key]; ok {
+		return
+	}
+
+	go job.run()
+
+	s.activeJobs.jobs[job.key] = job
+}
+
+// pollForNewScheduledJobs will only be started once per plugin. It doesn't need to be stopped.
+func (s *JobOnceScheduler) pollForNewScheduledJobs() {
+	for {
+		<-time.After(pollNewJobsInterval + addJitter())
+
+		if err := s.scheduleNewJobsFromDB(); err != nil {
+			s.pluginAPI.LogError("pluginAPI scheduleOnce poller encountered an error but is still polling", "error", err)
+		}
+	}
+}
+
+func (s *JobOnceScheduler) verifyCallbackExists() error {
+	s.storedCallback.mu.Lock()
+	defer s.storedCallback.mu.Unlock()
+
+	if s.storedCallback.callback == nil {
+		return errors.New("set callback before starting the scheduler")
+	}
+	return nil
+}

server/public/pluginapi/cluster/job_once_test.go

@@ -0,0 +1,685 @@
+package cluster
+
+import (
+	"encoding/json"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestScheduleOnceParallel(t *testing.T) {
+	makeKey := model.NewId
+
+	// there is only one callback by design, so all tests need to add their key
+	// and callback handling code here.
+	jobKey1 := makeKey()
+	count1 := new(int32)
+	jobKey2 := makeKey()
+	count2 := new(int32)
+	jobKey3 := makeKey()
+	jobKey4 := makeKey()
+	count4 := new(int32)
+	jobKey5 := makeKey()
+	count5 := new(int32)
+
+	manyJobs := make(map[string]*int32)
+	for i := 0; i < 100; i++ {
+		manyJobs[makeKey()] = new(int32)
+	}
+
+	callback := func(key string, _ any) {
+		switch key {
+		case jobKey1:
+			atomic.AddInt32(count1, 1)
+		case jobKey2:
+			atomic.AddInt32(count2, 1)
+		case jobKey3:
+			return // do nothing, like an error occurred in the plugin
+		case jobKey4:
+			atomic.AddInt32(count4, 1)
+		case jobKey5:
+			atomic.AddInt32(count5, 1)
+		default:
+			count, ok := manyJobs[key]
+			if ok {
+				atomic.AddInt32(count, 1)
+				return
+			}
+		}
+	}
+
+	mockPluginAPI := newMockPluginAPI(t)
+	getVal := func(key string) []byte {
+		data, _ := mockPluginAPI.KVGet(key)
+		return data
+	}
+
+	s := GetJobOnceScheduler(mockPluginAPI)
+
+	// should error if we try to start without callback
+	err := s.Start()
+	require.Error(t, err)
+
+	err = s.SetCallback(callback)
+	require.NoError(t, err)
+	err = s.Start()
+	require.NoError(t, err)
+
+	jobs, err := s.ListScheduledJobs()
+	require.NoError(t, err)
+	require.Empty(t, jobs)
+
+	t.Run("one scheduled job", func(t *testing.T) {
+		t.Parallel()
+
+		job, err2 := s.ScheduleOnce(jobKey1, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err2)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey1))
+
+		time.Sleep(200*time.Millisecond + scheduleOnceJitter)
+
+		assert.Empty(t, getVal(oncePrefix+jobKey1))
+		s.activeJobs.mu.RLock()
+		assert.Empty(t, s.activeJobs.jobs[jobKey1])
+		s.activeJobs.mu.RUnlock()
+
+		// It's okay to cancel jobs extra times, even if they're completed.
+		job.Cancel()
+		job.Cancel()
+		job.Cancel()
+		job.Cancel()
+
+		// Should have been called once
+		assert.Equal(t, int32(1), atomic.LoadInt32(count1))
+	})
+
+	t.Run("one job, stopped before firing", func(t *testing.T) {
+		t.Parallel()
+
+		job, err2 := s.ScheduleOnce(jobKey2, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err2)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey2))
+
+		job.Cancel()
+		assert.Empty(t, getVal(oncePrefix+jobKey2))
+		s.activeJobs.mu.RLock()
+		assert.Empty(t, s.activeJobs.jobs[jobKey2])
+		s.activeJobs.mu.RUnlock()
+
+		time.Sleep(2 * (waitAfterFail + scheduleOnceJitter))
+
+		// Should not have been called
+		assert.Equal(t, int32(0), atomic.LoadInt32(count2))
+
+		// It's okay to cancel jobs extra times, even if they're completed.
+		job.Cancel()
+		job.Cancel()
+		job.Cancel()
+		job.Cancel()
+	})
+
+	t.Run("failed at the plugin, job removed from db", func(t *testing.T) {
+		t.Parallel()
+
+		job, err2 := s.ScheduleOnce(jobKey3, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err2)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey3))
+
+		time.Sleep(200*time.Millisecond + scheduleOnceJitter)
+		assert.Empty(t, getVal(oncePrefix+jobKey3))
+		s.activeJobs.mu.RLock()
+		assert.Empty(t, s.activeJobs.jobs[jobKey3])
+		s.activeJobs.mu.RUnlock()
+	})
+
+	t.Run("cancel and restart a job with the same key", func(t *testing.T) {
+		t.Parallel()
+
+		job, err2 := s.ScheduleOnce(jobKey4, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err2)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey4))
+
+		job.Cancel()
+		assert.Empty(t, getVal(oncePrefix+jobKey4))
+		s.activeJobs.mu.RLock()
+		assert.Empty(t, s.activeJobs.jobs[jobKey4])
+		s.activeJobs.mu.RUnlock()
+
+		job, err2 = s.ScheduleOnce(jobKey4, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err2)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey4))
+
+		time.Sleep(200*time.Millisecond + scheduleOnceJitter)
+		assert.Equal(t, int32(1), atomic.LoadInt32(count4))
+		assert.Empty(t, getVal(oncePrefix+jobKey4))
+		s.activeJobs.mu.RLock()
+		assert.Empty(t, s.activeJobs.jobs[jobKey4])
+		s.activeJobs.mu.RUnlock()
+	})
+
+	t.Run("many scheduled jobs", func(t *testing.T) {
+		t.Parallel()
+
+		for k := range manyJobs {
+			job, err2 := s.ScheduleOnce(k, time.Now().Add(100*time.Millisecond), nil)
+			require.NoError(t, err2)
+			require.NotNil(t, job)
+			assert.NotEmpty(t, getVal(oncePrefix+k))
+		}
+
+		time.Sleep(200*time.Millisecond + scheduleOnceJitter)
+
+		for k, v := range manyJobs {
+			assert.Empty(t, getVal(oncePrefix+k))
+			s.activeJobs.mu.RLock()
+			assert.Empty(t, s.activeJobs.jobs[k])
+			s.activeJobs.mu.RUnlock()
+			assert.Equal(t, int32(1), *v)
+		}
+	})
+
+	t.Run("cancel a job by key name", func(t *testing.T) {
+		t.Parallel()
+
+		job, err2 := s.ScheduleOnce(jobKey5, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err2)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey5))
+		s.activeJobs.mu.RLock()
+		assert.NotEmpty(t, s.activeJobs.jobs[jobKey5])
+		s.activeJobs.mu.RUnlock()
+
+		s.Cancel(jobKey5)
+
+		assert.Empty(t, getVal(oncePrefix+jobKey5))
+		s.activeJobs.mu.RLock()
+		assert.Empty(t, s.activeJobs.jobs[jobKey5])
+		s.activeJobs.mu.RUnlock()
+
+		// cancel it again doesn't do anything:
+		s.Cancel(jobKey5)
+
+		time.Sleep(150*time.Millisecond + scheduleOnceJitter)
+		assert.Equal(t, int32(0), atomic.LoadInt32(count5))
+	})
+
+	t.Run("starting the scheduler again will return an error", func(t *testing.T) {
+		t.Parallel()
+
+		newScheduler := GetJobOnceScheduler(mockPluginAPI)
+		err = newScheduler.Start()
+		require.Error(t, err)
+	})
+}
+
+func TestScheduleOnceSequential(t *testing.T) {
+	makeKey := model.NewId
+
+	// get the existing scheduler
+	s := GetJobOnceScheduler(newMockPluginAPI(t))
+	getVal := func(key string) []byte {
+		data, _ := s.pluginAPI.KVGet(key)
+		return data
+	}
+	setMetadata := func(key string, metadata JobOnceMetadata) error {
+		data, err := json.Marshal(metadata)
+		if err != nil {
+			return err
+		}
+		ok, appErr := s.pluginAPI.KVSetWithOptions(oncePrefix+key, data, model.PluginKVSetOptions{})
+		if !ok {
+			return errors.New("KVSetWithOptions failed")
+		}
+		if appErr != nil {
+			return normalizeAppErr(appErr)
+		}
+		return nil
+	}
+
+	resetScheduler := func() {
+		s.activeJobs.mu.Lock()
+		defer s.activeJobs.mu.Unlock()
+		s.activeJobs.jobs = make(map[string]*JobOnce)
+		s.storedCallback.mu.Lock()
+		defer s.storedCallback.mu.Unlock()
+		s.storedCallback.callback = nil
+		s.startedMu.Lock()
+		defer s.startedMu.Unlock()
+		s.started = false
+		s.pluginAPI.(*mockPluginAPI).clear()
+	}
+
+	t.Run("starting the scheduler without a callback will return an error", func(t *testing.T) {
+		resetScheduler()
+
+		err := s.Start()
+		require.Error(t, err)
+	})
+
+	t.Run("trying to schedule a job without starting will return an error", func(t *testing.T) {
+		resetScheduler()
+
+		callback := func(key string, _ any) {}
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+
+		_, err = s.ScheduleOnce("will fail", time.Now(), nil)
+		require.Error(t, err)
+	})
+
+	t.Run("adding two callback works, only second one is called", func(t *testing.T) {
+		resetScheduler()
+
+		newCount2 := new(int32)
+		newCount3 := new(int32)
+
+		callback2 := func(key string, _ any) {
+			atomic.AddInt32(newCount2, 1)
+		}
+		callback3 := func(key string, _ any) {
+			atomic.AddInt32(newCount3, 1)
+		}
+
+		err := s.SetCallback(callback2)
+		require.NoError(t, err)
+		err = s.SetCallback(callback3)
+		require.NoError(t, err)
+		err = s.Start()
+		require.NoError(t, err)
+
+		_, err = s.ScheduleOnce("anything", time.Now().Add(50*time.Millisecond), nil)
+		require.NoError(t, err)
+		time.Sleep(70*time.Millisecond + scheduleOnceJitter)
+		assert.Equal(t, int32(0), atomic.LoadInt32(newCount2))
+		assert.Equal(t, int32(1), atomic.LoadInt32(newCount3))
+	})
+
+	t.Run("test paging keys from the db by inserting 3 pages of jobs and starting scheduler", func(t *testing.T) {
+		resetScheduler()
+
+		numPagingJobs := keysPerPage*3 + 2
+		testPagingJobs := make(map[string]*int32)
+		for i := 0; i < numPagingJobs; i++ {
+			testPagingJobs[makeKey()] = new(int32)
+		}
+
+		callback := func(key string, _ any) {
+			count, ok := testPagingJobs[key]
+			if ok {
+				atomic.AddInt32(count, 1)
+				return
+			}
+		}
+
+		// add the test paging jobs before starting scheduler
+		for k := range testPagingJobs {
+			assert.Empty(t, getVal(oncePrefix+k))
+			job, err := newJobOnce(s.pluginAPI, k, time.Now().Add(100*time.Millisecond), s.storedCallback, s.activeJobs, nil)
+			require.NoError(t, err)
+			err = job.saveMetadata()
+			require.NoError(t, err)
+			assert.NotEmpty(t, getVal(oncePrefix+k))
+		}
+
+		jobs, err := s.ListScheduledJobs()
+		require.NoError(t, err)
+		assert.Equal(t, len(testPagingJobs), len(jobs))
+
+		err = s.SetCallback(callback)
+		require.NoError(t, err)
+
+		//  reschedule from the db:
+		err = s.scheduleNewJobsFromDB()
+		require.NoError(t, err)
+
+		// wait for the testPagingJobs created in the setup to finish
+		time.Sleep(300 * time.Millisecond)
+
+		numInDB := 0
+		numActive := 0
+		numCountsAtZero := 0
+		for k, v := range testPagingJobs {
+			if getVal(oncePrefix+k) != nil {
+				numInDB++
+			}
+			s.activeJobs.mu.RLock()
+			if s.activeJobs.jobs[k] != nil {
+				numActive++
+			}
+			s.activeJobs.mu.RUnlock()
+			if atomic.LoadInt32(v) == int32(0) {
+				numCountsAtZero++
+			}
+		}
+
+		assert.Equal(t, 0, numInDB)
+		assert.Equal(t, 0, numActive)
+		assert.Equal(t, 0, numCountsAtZero)
+	})
+
+	t.Run("failed at the db", func(t *testing.T) {
+		resetScheduler()
+
+		jobKey1 := makeKey()
+		count1 := new(int32)
+
+		callback := func(key string, _ any) {
+			if key == jobKey1 {
+				atomic.AddInt32(count1, 1)
+			}
+		}
+
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		err = s.Start()
+		require.NoError(t, err)
+
+		jobs, err := s.ListScheduledJobs()
+		require.NoError(t, err)
+		require.Empty(t, jobs)
+
+		job, err := s.ScheduleOnce(jobKey1, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey1))
+		assert.NotEmpty(t, s.activeJobs.jobs[jobKey1])
+		s.pluginAPI.(*mockPluginAPI).setFailingWithPrefix(oncePrefix)
+
+		// wait until the metadata has failed to read
+		time.Sleep((maxNumFails + 1) * (waitAfterFail + scheduleOnceJitter))
+		assert.Equal(t, int32(0), atomic.LoadInt32(count1))
+		assert.Nil(t, getVal(oncePrefix+jobKey1))
+
+		assert.Empty(t, s.activeJobs.jobs[jobKey1])
+		assert.Empty(t, getVal(oncePrefix+jobKey1))
+		assert.Equal(t, int32(0), atomic.LoadInt32(count1))
+
+		s.pluginAPI.(*mockPluginAPI).setFailingWithPrefix("")
+	})
+
+	t.Run("simulate starting the plugin with 3 pending jobs in the db", func(t *testing.T) {
+		resetScheduler()
+
+		jobKeys := make(map[string]*int32)
+		for i := 0; i < 3; i++ {
+			jobKeys[makeKey()] = new(int32)
+		}
+
+		callback := func(key string, _ any) {
+			count, ok := jobKeys[key]
+			if ok {
+				atomic.AddInt32(count, 1)
+			}
+		}
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		err = s.Start()
+		require.NoError(t, err)
+
+		for k := range jobKeys {
+			job, err3 := newJobOnce(s.pluginAPI, k, time.Now().Add(100*time.Millisecond), s.storedCallback, s.activeJobs, nil)
+			require.NoError(t, err3)
+			err3 = job.saveMetadata()
+			require.NoError(t, err3)
+			assert.NotEmpty(t, getVal(oncePrefix+k))
+		}
+
+		// double checking they're in the db:
+		jobs, err := s.ListScheduledJobs()
+		require.NoError(t, err)
+		require.Len(t, jobs, 3)
+
+		// simulate starting the plugin
+		require.NoError(t, err)
+		err = s.scheduleNewJobsFromDB()
+		require.NoError(t, err)
+
+		time.Sleep(120*time.Millisecond + scheduleOnceJitter)
+
+		for k, v := range jobKeys {
+			assert.Empty(t, getVal(oncePrefix+k))
+			assert.Empty(t, s.activeJobs.jobs[k])
+			assert.Equal(t, int32(1), *v)
+		}
+		jobs, err = s.ListScheduledJobs()
+		require.NoError(t, err)
+		require.Empty(t, jobs)
+	})
+
+	t.Run("starting a job and polling before it's finished results in only one job running", func(t *testing.T) {
+		resetScheduler()
+
+		jobKey := makeKey()
+		count := new(int32)
+
+		callback := func(key string, _ any) {
+			if key == jobKey {
+				atomic.AddInt32(count, 1)
+			}
+		}
+
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		err = s.Start()
+		require.NoError(t, err)
+
+		jobs, err := s.ListScheduledJobs()
+		require.NoError(t, err)
+		require.Empty(t, jobs)
+
+		job, err := s.ScheduleOnce(jobKey, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey))
+		s.activeJobs.mu.Lock()
+		assert.NotEmpty(t, s.activeJobs.jobs[jobKey])
+		assert.Len(t, s.activeJobs.jobs, 1)
+		s.activeJobs.mu.Unlock()
+
+		// simulate what the polling function will do for a long running job:
+		err = s.scheduleNewJobsFromDB()
+		require.NoError(t, err)
+		err = s.scheduleNewJobsFromDB()
+		require.NoError(t, err)
+		err = s.scheduleNewJobsFromDB()
+		require.NoError(t, err)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey))
+		s.activeJobs.mu.Lock()
+		assert.NotEmpty(t, s.activeJobs.jobs[jobKey])
+		assert.Len(t, s.activeJobs.jobs, 1)
+		s.activeJobs.mu.Unlock()
+
+		// now wait for it to complete
+		time.Sleep(120*time.Millisecond + scheduleOnceJitter)
+		assert.Equal(t, int32(1), atomic.LoadInt32(count))
+		assert.Empty(t, getVal(oncePrefix+jobKey))
+		s.activeJobs.mu.Lock()
+		assert.Empty(t, s.activeJobs.jobs)
+		s.activeJobs.mu.Unlock()
+	})
+
+	t.Run("starting the same job again while it's still active will fail", func(t *testing.T) {
+		resetScheduler()
+
+		jobKey := makeKey()
+		count := new(int32)
+
+		callback := func(key string, _ any) {
+			if key == jobKey {
+				atomic.AddInt32(count, 1)
+			}
+		}
+
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		err = s.Start()
+		require.NoError(t, err)
+
+		jobs, err := s.ListScheduledJobs()
+		require.NoError(t, err)
+		require.Empty(t, jobs)
+
+		job, err := s.ScheduleOnce(jobKey, time.Now().Add(100*time.Millisecond), nil)
+		require.NoError(t, err)
+		require.NotNil(t, job)
+		assert.NotEmpty(t, getVal(oncePrefix+jobKey))
+		assert.NotEmpty(t, s.activeJobs.jobs[jobKey])
+		assert.Len(t, s.activeJobs.jobs, 1)
+
+		// a plugin tries to start the same jobKey again:
+		job, err = s.ScheduleOnce(jobKey, time.Now().Add(10000*time.Millisecond), nil)
+		require.Error(t, err)
+		require.Nil(t, job)
+
+		// now wait for first job to complete
+		time.Sleep(120*time.Millisecond + scheduleOnceJitter)
+		assert.Equal(t, int32(1), atomic.LoadInt32(count))
+		assert.Empty(t, getVal(oncePrefix+jobKey))
+		assert.Empty(t, s.activeJobs.jobs)
+	})
+
+	t.Run("simulate HA: canceling and setting a job with a different time--old one shouldn't fire", func(t *testing.T) {
+		resetScheduler()
+
+		key := makeKey()
+		jobKeys := make(map[string]*int32)
+		jobKeys[key] = new(int32)
+
+		// control is like the "control group" in an experiment. It will be overwritten,
+		// but with the same runAt. It should fire.
+		control := makeKey()
+		jobKeys[control] = new(int32)
+
+		callback := func(key string, _ any) {
+			count, ok := jobKeys[key]
+			if ok {
+				atomic.AddInt32(count, 1)
+			}
+		}
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		err = s.Start()
+		require.NoError(t, err)
+
+		originalRunAt := time.Now().Add(100 * time.Millisecond)
+		newRunAt := time.Now().Add(101 * time.Millisecond)
+
+		// store original
+		job, err := newJobOnce(s.pluginAPI, key, originalRunAt, s.storedCallback, s.activeJobs, nil)
+		require.NoError(t, err)
+		err = job.saveMetadata()
+		require.NoError(t, err)
+		assert.NotEmpty(t, getVal(oncePrefix+key))
+
+		// store oringal control
+		job2, err := newJobOnce(s.pluginAPI, control, originalRunAt, s.storedCallback, s.activeJobs, nil)
+		require.NoError(t, err)
+		err = job2.saveMetadata()
+		require.NoError(t, err)
+		assert.NotEmpty(t, getVal(oncePrefix+control))
+
+		// double checking originals are in the db:
+		jobs, err := s.ListScheduledJobs()
+		require.NoError(t, err)
+		require.Len(t, jobs, 2)
+		require.True(t, originalRunAt.Equal(jobs[0].RunAt))
+		require.True(t, originalRunAt.Equal(jobs[1].RunAt))
+
+		// simulate starting the plugin
+		require.NoError(t, err)
+		err = s.scheduleNewJobsFromDB()
+		require.NoError(t, err)
+
+		// Now "cancel" the original and make a new job with the same key but a different time.
+		// However, because we have only one list of synced jobs, we can't make two jobs with the
+		// same key. So we'll simulate this by changing the job metadata in the db. When the original
+		// job fires, it should see that the runAt is different, and it will think it has been canceled.
+		err = setMetadata(key, JobOnceMetadata{
+			Key:   key,
+			RunAt: newRunAt,
+		})
+		require.NoError(t, err)
+
+		// overwrite the control with the same runAt. It should fire.
+		err = setMetadata(control, JobOnceMetadata{
+			Key:   control,
+			RunAt: originalRunAt,
+		})
+		require.NoError(t, err)
+
+		time.Sleep(120*time.Millisecond + scheduleOnceJitter)
+
+		// original job didn't fire the callback:
+		assert.Empty(t, getVal(oncePrefix+key))
+		assert.Empty(t, s.activeJobs.jobs[key])
+		assert.Equal(t, int32(0), *jobKeys[key])
+
+		// control job did fire the callback:
+		assert.Empty(t, getVal(oncePrefix+control))
+		assert.Empty(t, s.activeJobs.jobs[control])
+		assert.Equal(t, int32(1), *jobKeys[control])
+
+		jobs, err = s.ListScheduledJobs()
+		require.NoError(t, err)
+		require.Empty(t, jobs)
+	})
+}
+
+func TestScheduleOnceProps(t *testing.T) {
+	t.Run("confirm props are returned", func(t *testing.T) {
+		s := GetJobOnceScheduler(newMockPluginAPI(t))
+
+		jobKey := model.NewId()
+		jobProps := struct {
+			Foo string
+		}{
+			Foo: "some foo",
+		}
+
+		var mut sync.Mutex
+		var called bool
+		callback := func(key string, props any) {
+			require.Equal(t, jobKey, key)
+			require.Equal(t, jobProps, props)
+			mut.Lock()
+			defer mut.Unlock()
+			called = true
+		}
+
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		if !s.started {
+			err = s.Start()
+			require.NoError(t, err)
+		}
+
+		_, err = s.ScheduleOnce(jobKey, time.Now().Add(100*time.Millisecond), jobProps)
+		require.NoError(t, err)
+
+		// Check if callback was called
+		require.Eventually(t, func() bool { mut.Lock(); defer mut.Unlock(); return called }, time.Second, 50*time.Millisecond)
+	})
+
+	t.Run("props to large", func(t *testing.T) {
+		s := GetJobOnceScheduler(newMockPluginAPI(t))
+
+		props := make([]byte, propsLimit)
+		for i := 0; i < propsLimit; i++ {
+			props[i] = 'a'
+		}
+
+		_, err := s.ScheduleOnce(model.NewId(), time.Now().Add(100*time.Millisecond), props)
+		require.Error(t, err)
+	})
+}

server/public/pluginapi/cluster/job_test.go

@@ -0,0 +1,412 @@
+package cluster
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMakeWaitForInterval(t *testing.T) {
+	t.Run("panics on invalid interval", func(t *testing.T) {
+		assert.Panics(t, func() {
+			MakeWaitForInterval(0)
+		})
+	})
+
+	const neverRun = -1 * time.Second
+
+	testCases := []struct {
+		Description  string
+		Interval     time.Duration
+		LastFinished time.Duration
+		Expected     time.Duration
+	}{
+		{
+			"never run, 5 minutes",
+			5 * time.Minute,
+			neverRun,
+			0,
+		},
+		{
+			"run 1 minute ago, 5 minutes",
+			5 * time.Minute,
+			-1 * time.Minute,
+			4 * time.Minute,
+		},
+		{
+			"run 2 minutes ago, 5 minutes",
+			5 * time.Minute,
+			-2 * time.Minute,
+			3 * time.Minute,
+		},
+		{
+			"run 4 minutes 30 seconds ago, 5 minutes",
+			5 * time.Minute,
+			-4*time.Minute - 30*time.Second,
+			30 * time.Second,
+		},
+		{
+			"run 4 minutes 59 seconds ago, 5 minutes",
+			5 * time.Minute,
+			-4*time.Minute - 59*time.Second,
+			1 * time.Second,
+		},
+		{
+			"never run, 1 hour",
+			1 * time.Hour,
+			neverRun,
+			0,
+		},
+		{
+			"run 1 minute ago, 1 hour",
+			1 * time.Hour,
+			-1 * time.Minute,
+			59 * time.Minute,
+		},
+		{
+			"run 20 minutes ago, 1 hour",
+			1 * time.Hour,
+			-20 * time.Minute,
+			40 * time.Minute,
+		},
+		{
+			"run 55 minutes 30 seconds ago, 1 hour",
+			1 * time.Hour,
+			-55*time.Minute - 30*time.Second,
+			4*time.Minute + 30*time.Second,
+		},
+		{
+			"run 59 minutes 59 seconds ago, 1 hour",
+			1 * time.Hour,
+			-59*time.Minute - 59*time.Second,
+			1 * time.Second,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.Description, func(t *testing.T) {
+			now := time.Now()
+
+			var lastFinished time.Time
+			if testCase.LastFinished != neverRun {
+				lastFinished = now.Add(testCase.LastFinished)
+			}
+
+			actual := MakeWaitForInterval(testCase.Interval)(now, JobMetadata{
+				LastFinished: lastFinished,
+			})
+			assert.Equal(t, testCase.Expected, actual)
+		})
+	}
+}
+
+func TestMakeWaitForRoundedInterval(t *testing.T) {
+	t.Run("panics on invalid interval", func(t *testing.T) {
+		assert.Panics(t, func() {
+			MakeWaitForRoundedInterval(0)
+		})
+	})
+
+	const neverRun = -1 * time.Second
+	topOfTheHour := time.Now().Truncate(1 * time.Hour)
+	topOfTheDay := time.Now().Truncate(24 * time.Hour)
+
+	testCases := []struct {
+		Description  string
+		Interval     time.Duration
+		Now          time.Time
+		LastFinished time.Duration
+		Expected     time.Duration
+	}{
+		{
+			"5 minutes, top of the hour, never run",
+			5 * time.Minute,
+			topOfTheHour,
+			neverRun,
+			0,
+		},
+		{
+			"5 minutes, top of the hour less 1 minute, never run",
+			5 * time.Minute,
+			topOfTheHour.Add(-1 * time.Minute),
+			neverRun,
+			0,
+		},
+		{
+			"5 minutes, top of the hour less 1 minute, run 1 minute ago",
+			5 * time.Minute,
+			topOfTheHour.Add(-1 * time.Minute),
+			-1 * time.Minute,
+			1 * time.Minute,
+		},
+		{
+			"5 minutes, top of the hour plus 1 minute, run 2 minutes ago",
+			5 * time.Minute,
+			topOfTheHour.Add(1 * time.Minute),
+			-2 * time.Minute,
+			0,
+		},
+		{
+			"5 minutes, top of the hour plus 1 minute, run 30 seconds ago",
+			5 * time.Minute,
+			topOfTheHour.Add(1 * time.Minute),
+			-30 * time.Second,
+			4 * time.Minute,
+		},
+		{
+			"5 minutes, top of the hour plus 7 minutes, run 30 seconds ago",
+			5 * time.Minute,
+			topOfTheHour.Add(7 * time.Minute),
+			-30 * time.Second,
+			3 * time.Minute,
+		},
+		{
+			"30 minutes, top of the hour, never run",
+			30 * time.Minute,
+			topOfTheHour,
+			neverRun,
+			0,
+		},
+		{
+			"30 minutes, top of the hour less 1 minute, never run",
+			30 * time.Minute,
+			topOfTheHour.Add(-1 * time.Minute),
+			neverRun,
+			0,
+		},
+		{
+			"30 minutes, top of the hour less 1 minute, run 1 minute ago",
+			30 * time.Minute,
+			topOfTheHour.Add(-1 * time.Minute),
+			-1 * time.Minute,
+			1 * time.Minute,
+		},
+		{
+			"30 minutes, top of the hour plus 1 minute, run 2 minutes ago",
+			30 * time.Minute,
+			topOfTheHour.Add(1 * time.Minute),
+			-2 * time.Minute,
+			0,
+		},
+		{
+			"30 minutes, top of the hour plus 1 minute, run 30 seconds ago",
+			30 * time.Minute,
+			topOfTheHour.Add(1 * time.Minute),
+			-30 * time.Second,
+			29 * time.Minute,
+		},
+		{
+			"30 minutes, top of the hour plus 7 minutes, run 30 seconds ago",
+			30 * time.Minute,
+			topOfTheHour.Add(7 * time.Minute),
+			-30 * time.Second,
+			23 * time.Minute,
+		},
+		{
+			"24 hours, top of the day, never run",
+			24 * time.Hour,
+			topOfTheDay,
+			neverRun,
+			0,
+		},
+		{
+			"24 hours, top of the day less 1 minute, never run",
+			24 * time.Hour,
+			topOfTheDay.Add(-1 * time.Minute),
+			neverRun,
+			0,
+		},
+		{
+			"24 hours, top of the day less 1 minute, run 1 minute ago",
+			24 * time.Hour,
+			topOfTheDay.Add(-1 * time.Minute),
+			-1 * time.Minute,
+			1 * time.Minute,
+		},
+		{
+			"24 hours, top of the day plus 1 minute, run 2 minutes ago",
+			24 * time.Hour,
+			topOfTheDay.Add(1 * time.Minute),
+			-2 * time.Minute,
+			0,
+		},
+		{
+			"24 hours, top of the day plus 1 minute, run 30 seconds ago",
+			24 * time.Hour,
+			topOfTheDay.Add(1 * time.Minute),
+			-30 * time.Second,
+			23*time.Hour + 59*time.Minute,
+		},
+		{
+			"24 hours, top of the day plus 7 minutes, run 30 seconds ago",
+			24 * time.Hour,
+			topOfTheDay.Add(7 * time.Minute),
+			-30 * time.Second,
+			23*time.Hour + 53*time.Minute,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.Description, func(t *testing.T) {
+			var lastFinished time.Time
+			if testCase.LastFinished != neverRun {
+				lastFinished = testCase.Now.Add(testCase.LastFinished)
+			}
+
+			actual := MakeWaitForRoundedInterval(testCase.Interval)(testCase.Now, JobMetadata{
+				LastFinished: lastFinished,
+			})
+			assert.Equal(t, testCase.Expected, actual)
+		})
+	}
+}
+
+func TestSchedule(t *testing.T) {
+	t.Parallel()
+
+	makeKey := model.NewId
+
+	t.Run("single-threaded", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		count := new(int32)
+		callback := func() {
+			atomic.AddInt32(count, 1)
+		}
+
+		job, err := Schedule(mockPluginAPI, makeKey(), MakeWaitForInterval(100*time.Millisecond), callback)
+		require.NoError(t, err)
+		require.NotNil(t, job)
+
+		time.Sleep(1 * time.Second)
+
+		err = job.Close()
+		require.NoError(t, err)
+
+		time.Sleep(1 * time.Second)
+
+		// Shouldn't have hit 20 in this time frame
+		assert.Less(t, *count, int32(20))
+
+		// Should have hit at least 5 in this time frame
+		assert.Greater(t, *count, int32(5))
+	})
+
+	t.Run("multi-threaded, single job", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		count := new(int32)
+		callback := func() {
+			atomic.AddInt32(count, 1)
+		}
+
+		var jobs []*Job
+
+		key := makeKey()
+
+		for i := 0; i < 3; i++ {
+			job, err := Schedule(mockPluginAPI, key, MakeWaitForInterval(100*time.Millisecond), callback)
+			require.NoError(t, err)
+			require.NotNil(t, job)
+
+			jobs = append(jobs, job)
+		}
+
+		time.Sleep(1 * time.Second)
+
+		var wg sync.WaitGroup
+		for i := 0; i < 3; i++ {
+			job := jobs[i]
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				err := job.Close()
+				require.NoError(t, err)
+			}()
+		}
+		wg.Wait()
+
+		time.Sleep(1 * time.Second)
+
+		// Shouldn't have hit 20 in this time frame
+		assert.Less(t, *count, int32(20))
+
+		// Should have hit at least 5 in this time frame
+		assert.Greater(t, *count, int32(5))
+	})
+
+	t.Run("multi-threaded, multiple jobs", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		countA := new(int32)
+		callbackA := func() {
+			atomic.AddInt32(countA, 1)
+		}
+
+		countB := new(int32)
+		callbackB := func() {
+			atomic.AddInt32(countB, 1)
+		}
+
+		keyA := makeKey()
+		keyB := makeKey()
+
+		var jobs []*Job
+		for i := 0; i < 3; i++ {
+			var key string
+			var callback func()
+			if i <= 1 {
+				key = keyA
+				callback = callbackA
+			} else {
+				key = keyB
+				callback = callbackB
+			}
+
+			job, err := Schedule(mockPluginAPI, key, MakeWaitForInterval(100*time.Millisecond), callback)
+			require.NoError(t, err)
+			require.NotNil(t, job)
+
+			jobs = append(jobs, job)
+		}
+
+		time.Sleep(1 * time.Second)
+
+		var wg sync.WaitGroup
+		for i := 0; i < 3; i++ {
+			job := jobs[i]
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				err := job.Close()
+				require.NoError(t, err)
+			}()
+		}
+		wg.Wait()
+
+		time.Sleep(1 * time.Second)
+
+		// Shouldn't have hit 20 in this time frame
+		assert.Less(t, *countA, int32(20))
+
+		// Should have hit at least 5 in this time frame
+		assert.Greater(t, *countA, int32(5))
+
+		// Shouldn't have hit 20 in this time frame
+		assert.Less(t, *countB, int32(20))
+
+		// Should have hit at least 5 in this time frame
+		assert.Greater(t, *countB, int32(5))
+	})
+}

server/public/pluginapi/cluster/mock_plugin_api_test.go

@@ -0,0 +1,150 @@
+package cluster
+
+import (
+	"bytes"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+type mockPluginAPI struct {
+	t *testing.T
+
+	lock              sync.Mutex
+	keyValues         map[string][]byte
+	failing           bool
+	failingWithPrefix string
+}
+
+func newMockPluginAPI(t *testing.T) *mockPluginAPI {
+	return &mockPluginAPI{
+		t:         t,
+		keyValues: make(map[string][]byte),
+	}
+}
+
+func (pluginAPI *mockPluginAPI) setFailing(failing bool) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	pluginAPI.failing = failing
+}
+
+func (pluginAPI *mockPluginAPI) setFailingWithPrefix(prefix string) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	pluginAPI.failingWithPrefix = prefix
+}
+
+func (pluginAPI *mockPluginAPI) clear() {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	for k := range pluginAPI.keyValues {
+		delete(pluginAPI.keyValues, k)
+	}
+}
+
+func (pluginAPI *mockPluginAPI) KVGet(key string) ([]byte, *model.AppError) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return nil, &model.AppError{Message: "fake error"}
+	}
+
+	if pluginAPI.failingWithPrefix != "" && strings.HasPrefix(key, pluginAPI.failingWithPrefix) {
+		return nil, &model.AppError{Message: "fake error for prefix " + pluginAPI.failingWithPrefix}
+	}
+
+	return pluginAPI.keyValues[key], nil
+}
+
+func (pluginAPI *mockPluginAPI) KVDelete(key string) *model.AppError {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return &model.AppError{Message: "fake error"}
+	}
+
+	if pluginAPI.failingWithPrefix != "" && strings.HasPrefix(key, pluginAPI.failingWithPrefix) {
+		return &model.AppError{Message: "fake error for prefix " + pluginAPI.failingWithPrefix}
+	}
+
+	delete(pluginAPI.keyValues, key)
+
+	return nil
+}
+
+func (pluginAPI *mockPluginAPI) KVList(page, count int) ([]string, *model.AppError) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return nil, &model.AppError{Message: "fake error"}
+	}
+
+	keys := make([]string, 0, len(pluginAPI.keyValues))
+	for k := range pluginAPI.keyValues {
+		keys = append(keys, k)
+	}
+
+	// have to sort, because we're paging below
+	sort.Strings(keys)
+
+	start := min(page*count, len(keys))
+	end := min((page+1)*count, len(keys))
+	return keys[start:end], nil
+}
+
+func min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func (pluginAPI *mockPluginAPI) KVSetWithOptions(key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError) {
+	pluginAPI.lock.Lock()
+	defer pluginAPI.lock.Unlock()
+
+	if pluginAPI.failing {
+		return false, &model.AppError{Message: "fake error"}
+	}
+
+	if pluginAPI.failingWithPrefix != "" && strings.HasPrefix(key, pluginAPI.failingWithPrefix) {
+		return false, &model.AppError{Message: "fake error for prefix " + pluginAPI.failingWithPrefix}
+	}
+
+	if options.Atomic {
+		if actualValue := pluginAPI.keyValues[key]; !bytes.Equal(actualValue, options.OldValue) {
+			return false, nil
+		}
+	}
+
+	if value == nil {
+		delete(pluginAPI.keyValues, key)
+	} else {
+		pluginAPI.keyValues[key] = value
+	}
+
+	return true, nil
+}
+
+func (pluginAPI *mockPluginAPI) LogError(msg string, keyValuePairs ...interface{}) {
+	if pluginAPI.t == nil {
+		return
+	}
+
+	pluginAPI.t.Helper()
+
+	params := []interface{}{msg}
+	params = append(params, keyValuePairs...)
+
+	pluginAPI.t.Log(params...)
+}

server/public/pluginapi/cluster/mutex.go

@@ -0,0 +1,185 @@
+package cluster
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/pkg/errors"
+)
+
+const (
+	// mutexPrefix is used to namespace key values created for a mutex from other key values
+	// created by a plugin.
+	mutexPrefix = "mutex_"
+)
+
+const (
+	// ttl is the interval after which a locked mutex will expire unless refreshed
+	ttl = time.Second * 15
+
+	// refreshInterval is the interval on which the mutex will be refreshed when locked
+	refreshInterval = ttl / 2
+)
+
+// MutexPluginAPI is the plugin API interface required to manage mutexes.
+type MutexPluginAPI interface {
+	KVSetWithOptions(key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError)
+	LogError(msg string, keyValuePairs ...interface{})
+}
+
+// Mutex is similar to sync.Mutex, except usable by multiple plugin instances across a cluster.
+//
+// Internally, a mutex relies on an atomic key-value set operation as exposed by the Mattermost
+// plugin API.
+//
+// Mutexes with different names are unrelated. Mutexes with the same name from different plugins
+// are unrelated. Pick a unique name for each mutex your plugin requires.
+//
+// A Mutex must not be copied after first use.
+type Mutex struct {
+	pluginAPI MutexPluginAPI
+	key       string
+
+	// lock guards the variables used to manage the refresh task, and is not itself related to
+	// the cluster-wide lock.
+	lock        sync.Mutex
+	stopRefresh chan bool
+	refreshDone chan bool
+}
+
+// NewMutex creates a mutex with the given key name.
+//
+// Panics if key is empty.
+func NewMutex(pluginAPI MutexPluginAPI, key string) (*Mutex, error) {
+	key, err := makeLockKey(key)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Mutex{
+		pluginAPI: pluginAPI,
+		key:       key,
+	}, nil
+}
+
+// makeLockKey returns the prefixed key used to namespace mutex keys.
+func makeLockKey(key string) (string, error) {
+	if key == "" {
+		return "", errors.New("must specify valid mutex key")
+	}
+
+	return mutexPrefix + key, nil
+}
+
+// lock makes a single attempt to atomically lock the mutex, returning true only if successful.
+func (m *Mutex) tryLock() (bool, error) {
+	ok, err := m.pluginAPI.KVSetWithOptions(m.key, []byte{1}, model.PluginKVSetOptions{
+		Atomic:          true,
+		OldValue:        nil, // No existing key value.
+		ExpireInSeconds: int64(ttl / time.Second),
+	})
+	if err != nil {
+		return false, errors.Wrap(err, "failed to set mutex kv")
+	}
+
+	return ok, nil
+}
+
+// refreshLock rewrites the lock key value with a new expiry, returning true only if successful.
+func (m *Mutex) refreshLock() error {
+	ok, err := m.pluginAPI.KVSetWithOptions(m.key, []byte{1}, model.PluginKVSetOptions{
+		Atomic:          true,
+		OldValue:        []byte{1},
+		ExpireInSeconds: int64(ttl / time.Second),
+	})
+	if err != nil {
+		return errors.Wrap(err, "failed to refresh mutex kv")
+	} else if !ok {
+		return errors.New("unexpectedly failed to refresh mutex kv")
+	}
+
+	return nil
+}
+
+// Lock locks m. If the mutex is already locked by any plugin instance, including the current one,
+// the calling goroutine blocks until the mutex can be locked.
+func (m *Mutex) Lock() {
+	_ = m.LockWithContext(context.Background())
+}
+
+// LockWithContext locks m unless the context is canceled. If the mutex is already locked by any plugin
+// instance, including the current one, the calling goroutine blocks until the mutex can be locked,
+// or the context is canceled.
+//
+// The mutex is locked only if a nil error is returned.
+func (m *Mutex) LockWithContext(ctx context.Context) error {
+	var waitInterval time.Duration
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(waitInterval):
+		}
+
+		locked, err := m.tryLock()
+		if err != nil {
+			m.pluginAPI.LogError("failed to lock mutex", "err", err, "lock_key", m.key)
+			waitInterval = nextWaitInterval(waitInterval, err)
+			continue
+		} else if !locked {
+			waitInterval = nextWaitInterval(waitInterval, err)
+			continue
+		}
+
+		stop := make(chan bool)
+		done := make(chan bool)
+		go func() {
+			defer close(done)
+			t := time.NewTicker(refreshInterval)
+			for {
+				select {
+				case <-t.C:
+					err := m.refreshLock()
+					if err != nil {
+						m.pluginAPI.LogError("failed to refresh mutex", "err", err, "lock_key", m.key)
+						return
+					}
+				case <-stop:
+					return
+				}
+			}
+		}()
+
+		m.lock.Lock()
+		m.stopRefresh = stop
+		m.refreshDone = done
+		m.lock.Unlock()
+
+		return nil
+	}
+}
+
+// Unlock unlocks m. It is a run-time error if m is not locked on entry to Unlock.
+//
+// Just like sync.Mutex, a locked Lock is not associated with a particular goroutine or plugin
+// instance. It is allowed for one goroutine or plugin instance to lock a Lock and then arrange
+// for another goroutine or plugin instance to unlock it. In practice, ownership of the lock should
+// remain within a single plugin instance.
+func (m *Mutex) Unlock() {
+	m.lock.Lock()
+	if m.stopRefresh == nil {
+		m.lock.Unlock()
+		panic("mutex has not been acquired")
+	}
+
+	close(m.stopRefresh)
+	m.stopRefresh = nil
+	<-m.refreshDone
+	m.lock.Unlock()
+
+	// If an error occurs deleting, the mutex kv will still expire, allowing later retry.
+	_, _ = m.pluginAPI.KVSetWithOptions(m.key, nil, model.PluginKVSetOptions{})
+}

server/public/pluginapi/cluster/mutex_example_test.go

@@ -0,0 +1,20 @@
+package cluster_test
+
+import (
+	"github.com/mattermost/mattermost/server/public/plugin"
+	"github.com/mattermost/mattermost/server/public/pluginapi/cluster"
+)
+
+//nolint:staticcheck
+func ExampleMutex() {
+	// Use p.API from your plugin instead.
+	pluginAPI := plugin.API(nil)
+
+	m, err := cluster.NewMutex(pluginAPI, "key")
+	if err != nil {
+		panic(err)
+	}
+	m.Lock()
+	// critical section
+	m.Unlock()
+}

server/public/pluginapi/cluster/mutex_test.go

@@ -0,0 +1,276 @@
+package cluster
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+func mustNewMutex(pluginAPI MutexPluginAPI, key string) *Mutex {
+	m, err := NewMutex(pluginAPI, key)
+	if err != nil {
+		panic(err)
+	}
+
+	return m
+}
+
+func TestMakeLockKey(t *testing.T) {
+	t.Run("fails when empty", func(t *testing.T) {
+		key, err := makeLockKey("")
+		assert.Error(t, err)
+		assert.Empty(t, key)
+	})
+
+	t.Run("not-empty", func(t *testing.T) {
+		testCases := map[string]string{
+			"key":   mutexPrefix + "key",
+			"other": mutexPrefix + "other",
+		}
+
+		for key, expected := range testCases {
+			actual, err := makeLockKey(key)
+			require.NoError(t, err)
+			assert.Equal(t, expected, actual)
+		}
+	})
+}
+
+func lock(t *testing.T, m *Mutex) {
+	t.Helper()
+
+	done := make(chan bool)
+	go func() {
+		t.Helper()
+
+		defer close(done)
+		m.Lock()
+	}()
+
+	select {
+	case <-time.After(1 * time.Second):
+		require.Fail(t, "failed to lock mutex within 1 second")
+	case <-done:
+	}
+}
+
+func unlock(t *testing.T, m *Mutex, panics bool) {
+	t.Helper()
+
+	done := make(chan bool)
+	go func() {
+		t.Helper()
+
+		defer close(done)
+		if panics {
+			assert.Panics(t, m.Unlock)
+		} else {
+			assert.NotPanics(t, m.Unlock)
+		}
+	}()
+
+	select {
+	case <-time.After(1 * time.Second):
+		require.Fail(t, "failed to unlock mutex within 1 second")
+	case <-done:
+	}
+}
+
+func TestMutex(t *testing.T) {
+	t.Parallel()
+
+	makeKey := model.NewId
+
+	t.Run("successful lock/unlock cycle", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		m := mustNewMutex(mockPluginAPI, makeKey())
+		lock(t, m)
+		unlock(t, m, false)
+		lock(t, m)
+		unlock(t, m, false)
+	})
+
+	t.Run("unlock when not locked", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		m := mustNewMutex(mockPluginAPI, makeKey())
+		unlock(t, m, true)
+	})
+
+	t.Run("blocking lock", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		m := mustNewMutex(mockPluginAPI, makeKey())
+		lock(t, m)
+
+		done := make(chan bool)
+		go func() {
+			defer close(done)
+			m.Lock()
+		}()
+
+		select {
+		case <-time.After(1 * time.Second):
+		case <-done:
+			require.Fail(t, "second goroutine should not have locked")
+		}
+
+		unlock(t, m, false)
+
+		select {
+		case <-time.After(pollWaitInterval * 2):
+			require.Fail(t, "second goroutine should have locked")
+		case <-done:
+		}
+	})
+
+	t.Run("failed lock", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		m := mustNewMutex(mockPluginAPI, makeKey())
+
+		mockPluginAPI.setFailing(true)
+
+		done := make(chan bool)
+		go func() {
+			defer close(done)
+			m.Lock()
+		}()
+
+		select {
+		case <-time.After(5 * time.Second):
+		case <-done:
+			require.Fail(t, "goroutine should not have locked")
+		}
+
+		mockPluginAPI.setFailing(false)
+
+		select {
+		case <-time.After(15 * time.Second):
+			require.Fail(t, "goroutine should have locked")
+		case <-done:
+		}
+	})
+
+	t.Run("failed unlock", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		key := makeKey()
+		m := mustNewMutex(mockPluginAPI, key)
+		lock(t, m)
+
+		mockPluginAPI.setFailing(true)
+
+		unlock(t, m, false)
+
+		// Simulate expiry
+		mockPluginAPI.clear()
+		mockPluginAPI.setFailing(false)
+
+		lock(t, m)
+	})
+
+	t.Run("discrete keys", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		m1 := mustNewMutex(mockPluginAPI, makeKey())
+		lock(t, m1)
+
+		m2 := mustNewMutex(mockPluginAPI, makeKey())
+		lock(t, m2)
+
+		m3 := mustNewMutex(mockPluginAPI, makeKey())
+		lock(t, m3)
+
+		unlock(t, m1, false)
+		unlock(t, m3, false)
+
+		lock(t, m1)
+
+		unlock(t, m2, false)
+		unlock(t, m1, false)
+	})
+
+	t.Run("with uncancelled context", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		key := makeKey()
+		m := mustNewMutex(mockPluginAPI, key)
+
+		m.Lock()
+
+		ctx := context.Background()
+		done := make(chan bool)
+		go func() {
+			defer close(done)
+			err := m.LockWithContext(ctx)
+			require.Nil(t, err)
+		}()
+
+		select {
+		case <-time.After(ttl + pollWaitInterval*2):
+		case <-done:
+			require.Fail(t, "goroutine should not have locked")
+		}
+
+		m.Unlock()
+
+		select {
+		case <-time.After(pollWaitInterval * 2):
+			require.Fail(t, "goroutine should have locked after unlock")
+		case <-done:
+		}
+	})
+
+	t.Run("with canceled context", func(t *testing.T) {
+		t.Parallel()
+
+		mockPluginAPI := newMockPluginAPI(t)
+
+		m := mustNewMutex(mockPluginAPI, makeKey())
+
+		m.Lock()
+
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan bool)
+		go func() {
+			defer close(done)
+			err := m.LockWithContext(ctx)
+			require.NotNil(t, err)
+		}()
+
+		select {
+		case <-time.After(ttl + pollWaitInterval*2):
+		case <-done:
+			require.Fail(t, "goroutine should not have locked")
+		}
+
+		cancel()
+
+		select {
+		case <-time.After(pollWaitInterval * 2):
+			require.Fail(t, "goroutine should have aborted after cancellation")
+		case <-done:
+		}
+	})
+}

server/public/pluginapi/cluster/wait.go

@@ -0,0 +1,43 @@
+package cluster
+
+import (
+	"math/rand"
+	"time"
+)
+
+const (
+	// minWaitInterval is the minimum amount of time to wait between locking attempts
+	minWaitInterval = 1 * time.Second
+
+	// maxWaitInterval is the maximum amount of time to wait between locking attempts
+	maxWaitInterval = 5 * time.Minute
+
+	// pollWaitInterval is the usual time to wait between unsuccessful locking attempts
+	pollWaitInterval = 1 * time.Second
+
+	// jitterWaitInterval is the amount of jitter to add when waiting to avoid thundering herds
+	jitterWaitInterval = minWaitInterval / 2
+)
+
+// nextWaitInterval determines how long to wait until the next lock retry.
+func nextWaitInterval(lastWaitInterval time.Duration, err error) time.Duration {
+	nextWaitInterval := lastWaitInterval
+
+	if nextWaitInterval <= 0 {
+		nextWaitInterval = minWaitInterval
+	}
+
+	if err != nil {
+		nextWaitInterval *= 2
+		if nextWaitInterval > maxWaitInterval {
+			nextWaitInterval = maxWaitInterval
+		}
+	} else {
+		nextWaitInterval = pollWaitInterval
+	}
+
+	// Add some jitter to avoid unnecessary collision between competing plugin instances.
+	nextWaitInterval += time.Duration(rand.Int63n(int64(jitterWaitInterval)) - int64(jitterWaitInterval)/2)
+
+	return nextWaitInterval
+}

server/public/pluginapi/cluster/wait_test.go

@@ -0,0 +1,156 @@
+package cluster
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNextWaitInterval(t *testing.T) {
+	testCases := []struct {
+		Description      string
+		lastWaitInterval time.Duration
+		err              error
+		expectedRange    [2]time.Duration
+	}{
+		{
+			"0, no error",
+			0,
+			nil,
+			[2]time.Duration{
+				1*time.Second - jitterWaitInterval/2,
+				1*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"0, error",
+			0,
+			errors.New("test"),
+			[2]time.Duration{
+				2*time.Second - jitterWaitInterval/2,
+				2*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"negative, no error",
+			-100 * time.Second,
+			nil,
+			[2]time.Duration{
+				1*time.Second - jitterWaitInterval/2,
+				1*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"negative, error",
+			-100 * time.Second,
+			errors.New("test"),
+			[2]time.Duration{
+				2*time.Second - jitterWaitInterval/2,
+				2*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"1 second, no error",
+			1 * time.Second,
+			nil,
+			[2]time.Duration{
+				1*time.Second - jitterWaitInterval/2,
+				1*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"1 second, error",
+			1 * time.Second,
+			errors.New("test"),
+			[2]time.Duration{
+				2*time.Second - jitterWaitInterval/2,
+				2*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"10 seconds, no error",
+			10 * time.Second,
+			nil,
+			[2]time.Duration{
+				1*time.Second - jitterWaitInterval/2,
+				1*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"10 second, error",
+			10 * time.Second,
+			errors.New("test"),
+			[2]time.Duration{
+				20*time.Second - jitterWaitInterval/2,
+				20*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"4 minutes, no error",
+			4 * time.Minute,
+			nil,
+			[2]time.Duration{
+				1*time.Second - jitterWaitInterval/2,
+				1*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"4 minutes, error",
+			4 * time.Minute,
+			errors.New("test"),
+			[2]time.Duration{
+				5*time.Minute - jitterWaitInterval/2,
+				5*time.Minute + jitterWaitInterval/2,
+			},
+		},
+		{
+			"5 minutes, no error",
+			5 * time.Minute,
+			nil,
+			[2]time.Duration{
+				1*time.Second - jitterWaitInterval/2,
+				1*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"5 minutes, error",
+			5 * time.Minute,
+			errors.New("test"),
+			[2]time.Duration{
+				5*time.Minute - jitterWaitInterval/2,
+				5*time.Minute + jitterWaitInterval/2,
+			},
+		},
+		{
+			"10minutes, no error",
+			10 * time.Minute,
+			nil,
+			[2]time.Duration{
+				1*time.Second - jitterWaitInterval/2,
+				1*time.Second + jitterWaitInterval/2,
+			},
+		},
+		{
+			"10minutes, error",
+			10 * time.Minute,
+			errors.New("test"),
+			[2]time.Duration{
+				5*time.Minute - jitterWaitInterval/2,
+				5*time.Minute + jitterWaitInterval/2,
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.Description, func(t *testing.T) {
+			actualWaitInterval := nextWaitInterval(
+				testCase.lastWaitInterval,
+				testCase.err,
+			)
+			assert.GreaterOrEqual(t, int64(actualWaitInterval), int64(testCase.expectedRange[0]))
+			assert.LessOrEqual(t, int64(actualWaitInterval), int64(testCase.expectedRange[1]))
+		})
+	}
+}