Clarifications

src/github.com/matrix-org/dendrite/clientapi/sync/requestpool.go

@@ -1,7 +1,6 @@
 package sync
 
 import (
-	"fmt"
 	"net/http"
 	"strconv"
 	"sync"
@@ -117,7 +116,6 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request) util.JSONRespons
 func (rp *RequestPool) OnNewEvent(ev *gomatrixserverlib.Event, pos syncStreamPosition) {
 	// update the current position in a guard and then notify all /sync streams
 	rp.cond.L.Lock()
-	fmt.Println("OnNewEvent =>", ev.EventID(), " pos=", pos, " old_pos=", rp.currPos)
 	rp.currPos = pos
 	rp.cond.L.Unlock()
 

src/github.com/matrix-org/dendrite/common/consumers.go

@@ -67,6 +67,8 @@ func (c *ContinualConsumer) Start() error {
 	}
 	for _, offset := range storedOffsets {
 		// We've already processed events from this partition so advance the offset to where we got to.
+		// Offsets are provided with each message, so if we use the same offset on startup then we'll
+		// get the same message a 2nd time, so increment 1 to indicate the next offset.
 		offsets[offset.Partition] = 1 + offset.Offset
 	}