possible fix for a race condition in topic subscriptions

server/datamodel.go

@@ -1673,8 +1673,8 @@ func ErrLocked(id, topic string, ts time.Time) *ServerComMessage {
 	return ErrLockedExplicitTs(id, topic, ts, ts)
 }
 
-// ErrLockedReply operation rejected because the topic is being deleted with explicit server and
-// incoming request timestamps in response to a client request (503).
+// ErrLockedReply operation rejected because the topic is being deleted in response
+// to a client request (503).
 func ErrLockedReply(msg *ClientComMessage, ts time.Time) *ServerComMessage {
 	return ErrLockedExplicitTs(msg.Id, msg.Original, ts, msg.Timestamp)
 }

server/store/types/types.go

@@ -51,6 +51,8 @@ const (
 	ErrInvalidResponse = StoreError("invalid response")
 	// ErrRedirected means the subscription request was redirected to another topic.
 	ErrRedirected = StoreError("redirected")
+	// ErrAlreadySatisfied means the desired result, such as subscription, is already satisfied.
+	ErrAlreadySatisfied = StoreError("already satisfied")
 )
 
 // Uid is a database-specific record id, suitable to be used as a primary key.

server/topic.go

@@ -1330,6 +1330,12 @@ func (t *Topic) broadcastToSessions(msg *ServerComMessage) {
 func (t *Topic) subscriptionReply(asChan bool, msg *ClientComMessage) error {
 	// The topic is already initialized by the Hub
 
+	// Check if the session is already subscribed to topic.
+	if msg.sess.getSub(t.name) != nil {
+		msg.sess.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
+		return types.ErrAlreadySatisfied
+	}
+
 	msgsub := msg.Sub
 
 	// For newly created topics report topic creation time.

server/utils.go

@@ -271,6 +271,8 @@ func decodeStoreErrorExplicitTs(err error, id, topic string, serverTs, incomingR
 			errmsg = ErrInvalidResponse(id, topic, serverTs, incomingReqTs)
 		case types.ErrRedirected:
 			errmsg = InfoUseOther(id, topic, params["topic"].(string), serverTs, incomingReqTs)
+		case types.ErrAlreadySatisfied:
+			errmsg = InfoNoAction(id, topic, serverTs, incomingReqTs)
 		default:
 			errmsg = ErrUnknownExplicitTs(id, topic, serverTs, incomingReqTs)
 		}