fix closing conns with insufficient state in case of non-atomic publi…

…sh to Redis (#120)
centrifugal · May 15, 2020 · c50fecc · c50fecc
1 parent d24832c
commit c50fecc
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 5 deletions.
diff --git a/_examples/go.sum b/_examples/go.sum
@@ -11,8 +11,8 @@ github.com/bradfitz/gomemcache v0.0.0-20190329173943-551aad21a668/go.mod h1:H0wQ
 github.com/bradleypeabody/gorilla-sessions-memcache v0.0.0-20181103040241-659414f458e1/go.mod h1:dkChI7Tbtx7H1Tj7TqGSZMOeGpMP5gLHtjroHd4agiI=
 github.com/centrifugal/protocol v0.3.3 h1:GCNee3RFsjQu6SyKBX0Ir7ByUrp+Gw0MU/PsIc2CM2s=
 github.com/centrifugal/protocol v0.3.3/go.mod h1:2YbBCaDwQHl37ErRdMrKSj18X2yVvpkQYtSX6aVbe5A=
-github.com/cristalhq/jwt/v2 v2.0.0 h1:CxleHxkZQQ5J0siUQ2gwZrhAysmh8Ddh/R06AzCiYao=
-github.com/cristalhq/jwt/v2 v2.0.0/go.mod h1:nQT19GqJbrWubmI+ULE8PYsR1GCbwI5hAg1nG+9AbTg=
+github.com/cristalhq/jwt/v3 v3.0.0 h1:0PtBLOa6XEZj6powOaneZCxU3+yiFSYwVeL2wmbavnY=
+github.com/cristalhq/jwt/v3 v3.0.0/go.mod h1:XOnIXst8ozq/esy5N1XOlSyQqBd+84fxJ99FK+1jgL8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

diff --git a/_examples/redis_engine/main.go b/_examples/redis_engine/main.go
@@ -152,11 +152,11 @@ func main() {
 				time.Sleep(time.Second)
 				_, err := node.Publish("chat:"+strconv.Itoa(i), []byte("hello"))
 				if err != nil {
-					panic(err.Error())
+					log.Println(err.Error())
 				}
 				_, err = node.History("chat:" + strconv.Itoa(i))
 				if err != nil {
-					panic(err.Error())
+					log.Println(err.Error())
 				}
 			}
 		}(i)

diff --git a/changelog.md b/changelog.md
@@ -1,3 +1,8 @@
+v0.8.1
+======
+
+* Fix closing connections with `insufficient state` after publish when history recovery feature is on and `PublishOnHistoryAdd` is `false` in Redis Engine config. [#119](https://github.com/centrifugal/centrifuge/issues/119).
+
 v0.8.0
 ======
 

diff --git a/engine_redis.go b/engine_redis.go
@@ -1038,7 +1038,12 @@ func (s *shard) handleRedisClientMessage(eventHandler BrokerEventHandler, _ chan
 		if err != nil {
 			return err
 		}
-		pub.Offset = offset
+		if pub.Offset == 0 {
+			// When adding to history and publishing happens atomically in Redis Engine
+			// offset is prepended to Publication payload. In this case we should attach
+			// it to unmarshalled Publication.
+			pub.Offset = offset
+		}
 		_ = eventHandler.HandlePublication(push.Channel, &pub)
 	case protocol.PushTypeJoin:
 		var join protocol.Join