Fix potential races when accessing or updating call post (#583)

go.mod

@@ -13,6 +13,7 @@ require (
 require (
 	github.com/Masterminds/semver v1.5.0
 	github.com/gorilla/websocket v1.5.0
+	github.com/jmoiron/sqlx v1.3.5
 	github.com/mattermost/calls-offloader v0.5.0
 	github.com/mattermost/calls-recorder v0.6.0
 	github.com/mattermost/calls-transcriber v0.1.0

go.sum

@@ -126,6 +126,7 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
+github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-sql-driver/mysql v1.7.1 h1:lUIinVbN1DY0xBg0eMOzmmtGoHwWBbvnWubQUrtU8EI=
 github.com/go-sql-driver/mysql v1.7.1/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -259,6 +260,8 @@ github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANyt
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
 github.com/jhump/protoreflect v1.6.0 h1:h5jfMVslIg6l29nsMs0D8Wj17RDVdNYti0vDN/PZZoE=
 github.com/jhump/protoreflect v1.6.0/go.mod h1:eaTn3RZAmMBcV0fifFvlm6VHNz3wSkYyXYWUh7ymB74=
+github.com/jmoiron/sqlx v1.3.5 h1:vFFPA71p1o5gAeqtEAwLU4dnX2napprKtHr7PYIcN3g=
+github.com/jmoiron/sqlx v1.3.5/go.mod h1:nRVWtLre0KfCLJvgxzCsLVMogSvQ1zNJtpYr2Ccp0mQ=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
@@ -285,6 +288,7 @@ github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 h1:SOEGU9fKiNWd/HOJuq
 github.com/lann/builder v0.0.0-20180802200727-47ae307949d0/go.mod h1:dXGbAdH5GtBTC4WfIxhKZfyBF/HBFgRZSWwZ9g/He9o=
 github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 h1:P6pPBnrTSX3DEVR4fDembhRWSsG5rVo6hYhAB/ADZrk=
 github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0/go.mod h1:vmVJ0l/dxyfGW6FmdpVm2joNMFikkuWg0EoCKLGUMNw=
+github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/lunixbochs/vtclean v1.0.0/go.mod h1:pHhQNgMf3btfWnGBVipUOjRYhoOsdGqdm/+2c2E2WMI=
@@ -320,6 +324,8 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
 github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
 github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
 github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mattn/go-sqlite3 v1.14.6 h1:dNPt6NO46WmLVt2DLNpwczCmdV5boIZ6g/tlDrlRUbg=
+github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
 github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=

server/bot_api.go

@@ -189,10 +189,21 @@ func (p *Plugin) handleBotPostRecordings(w http.ResponseWriter, r *http.Request,
 		return
 	}
 
+	// Here we need to lock since we'll be reading and updating the call
+	// post, potentially concurrently with other events (e.g. call ending,
+	// transcribing job completing).
+	_, err := p.lockCall(callID)
+	if err != nil {
+		res.Err = fmt.Errorf("failed to lock call: %w", err).Error()
+		res.Code = http.StatusInternalServerError
+		return
+	}
+	defer p.unlockCall(callID)
+
 	// Update call post
-	post, appErr := p.API.GetPost(info.PostID)
-	if appErr != nil {
-		res.Err = "failed to get call post: " + appErr.Error()
+	post, err := p.GetPost(info.PostID)
+	if err != nil {
+		res.Err = "failed to get call post: " + err.Error()
 		res.Code = http.StatusInternalServerError
 		return
 	}
@@ -236,7 +247,7 @@ func (p *Plugin) handleBotPostRecordings(w http.ResponseWriter, r *http.Request,
 	recPost.AddProp("recording_id", info.JobID)
 	recPost.AddProp("call_post_id", info.PostID)
 
-	recPost, appErr = p.API.CreatePost(recPost)
+	recPost, appErr := p.API.CreatePost(recPost)
 	if appErr != nil {
 		res.Err = "failed to create post: " + appErr.Error()
 		res.Code = http.StatusInternalServerError
@@ -284,10 +295,21 @@ func (p *Plugin) handleBotPostTranscriptions(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
+	// Here we need to lock since we'll be reading and updating the call
+	// post, potentially concurrently with other events (e.g. call ending,
+	// recording job completing).
+	_, err := p.lockCall(callID)
+	if err != nil {
+		res.Err = fmt.Errorf("failed to lock call: %w", err).Error()
+		res.Code = http.StatusInternalServerError
+		return
+	}
+	defer p.unlockCall(callID)
+
 	// Update call post
-	post, appErr := p.API.GetPost(info.PostID)
-	if appErr != nil {
-		res.Err = "failed to get call post: " + appErr.Error()
+	post, err := p.GetPost(info.PostID)
+	if err != nil {
+		res.Err = "failed to get call post: " + err.Error()
 		res.Code = http.StatusInternalServerError
 		return
 	}
@@ -361,9 +383,9 @@ func (p *Plugin) handleBotPostTranscriptions(w http.ResponseWriter, r *http.Requ
 	var rm jobMetadata
 	rm.fromMap(recordings[tm.RecID])
 	if rm.PostID != "" {
-		recPost, appErr := p.API.GetPost(rm.PostID)
-		if appErr != nil {
-			res.Err = "failed to get recording post: " + appErr.Error()
+		recPost, err := p.GetPost(rm.PostID)
+		if err != nil {
+			res.Err = "failed to get recording post: " + err.Error()
 			res.Code = http.StatusInternalServerError
 			p.LogError(res.Err, "trID", info.JobID)
 			return

server/job_metadata.go

@@ -76,9 +76,9 @@ func (jm *jobMetadata) fromMap(data any) {
 }
 
 func (p *Plugin) saveRecordingMetadata(postID, recID, trID string) error {
-	post, appErr := p.API.GetPost(postID)
-	if appErr != nil {
-		return fmt.Errorf("failed to get call post: %w", appErr)
+	post, err := p.GetPost(postID)
+	if err != nil {
+		return fmt.Errorf("failed to get call post: %w", err)
 	}
 
 	rm := jobMetadata{

server/plugin.go

@@ -22,6 +22,8 @@ import (
 
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/plugin"
+
+	"github.com/jmoiron/sqlx"
 )
 
 const (
@@ -67,6 +69,7 @@ type Plugin struct {
 
 	// Database handle to the writer DB node
 	wDB        *sql.DB
+	wDBx       *sqlx.DB
 	driverName string
 }
 
@@ -338,9 +341,9 @@ func (p *Plugin) updateCallPostEnded(postID string, participants []string) (floa
 		return 0, fmt.Errorf("postID should not be empty")
 	}
 
-	post, appErr := p.API.GetPost(postID)
-	if appErr != nil {
-		return 0, appErr
+	post, err := p.GetPost(postID)
+	if err != nil {
+		return 0, err
 	}
 
 	postMsg := "Call ended"
@@ -356,8 +359,7 @@ func (p *Plugin) updateCallPostEnded(postID string, participants []string) (floa
 	post.AddProp("end_at", time.Now().UnixMilli())
 	post.AddProp("participants", participants)
 
-	_, appErr = p.API.UpdatePost(post)
-	if appErr != nil {
+	if _, appErr := p.API.UpdatePost(post); appErr != nil {
 		return 0, appErr
 	}
 

server/store.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/mattermost/mattermost/server/public/model"
 
+	"github.com/jmoiron/sqlx"
 	sq "github.com/mattermost/squirrel"
 )
 
@@ -29,6 +30,7 @@ func (p *Plugin) initDB() error {
 	if err := p.wDB.Ping(); err != nil {
 		return fmt.Errorf("failed to ping writer DB: %w", err)
 	}
+	p.wDBx = sqlx.NewDb(p.wDB, p.driverName)
 
 	p.LogInfo("handle to writer DB initialized successfully", "driver", p.driverName)
 
@@ -79,6 +81,31 @@ func (p *Plugin) KVGet(key string, fromWriter bool) ([]byte, error) {
 	return data, nil
 }
 
+// GetPost is an alternative to p.API.GetPost() that fetches from the writer DB node.
+// This should only be used internally to get calls posts as it doesn't take care of more
+// advanced logic needed by clients like populating reply counts.
+func (p *Plugin) GetPost(postID string) (*model.Post, error) {
+	p.metrics.IncStoreOp("GetPost")
+
+	qb := getQueryBuilder(p.driverName).
+		Select("*").
+		From("Posts").
+		Where(sq.Eq{"Id": postID})
+	q, args, err := qb.ToSql()
+	if err != nil {
+		return nil, fmt.Errorf("failed to prepare query: %w", err)
+	}
+
+	var post model.Post
+	if err := p.wDBx.Get(&post, q, args...); err == sql.ErrNoRows {
+		return nil, fmt.Errorf("post not found (id=%s)", postID)
+	} else if err != nil {
+		return nil, fmt.Errorf("failed to get post (id=%s): %w", postID, err)
+	}
+
+	return &post, nil
+}
+
 func (p *Plugin) updateFileInfoPostID(fileID, postID string) error {
 	qb := getQueryBuilder(p.driverName).Update("FileInfo").
 		Set("PostId", postID).