more sync api bug fixes

internal/api/syncapi/authentication.go

@@ -39,17 +39,14 @@ func ContextWithPeer(ctx context.Context, peer *v1.Multihost_Peer, publicKey *cr
 // HTTP decorator for authentication middleware.
 func AuthenticationMiddleware(configManager *config.ConfigManager, handler http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		zap.S().Debugf("AuthenticationMiddleware called for %s %s", r.Method, r.URL.Path)
 		authHeader := r.Header.Get("Authorization")
 		if authHeader == "" {
-			zap.S().Error("missing Authorization header in request")
 			http.Error(w, "Unauthorized: missing authentication header", http.StatusUnauthorized)
 			return
 		}
 
 		config, err := configManager.Get()
 		if err != nil {
-			zap.S().Errorf("failed to get authorized clients from config: %v", err)
 			http.Error(w, "Internal error", http.StatusInternalServerError)
 			return
 		}
@@ -57,7 +54,6 @@ func AuthenticationMiddleware(configManager *config.ConfigManager, handler http.
 
 		peerKey, instanceID, err := verifyAuthenticationHeader(authHeader)
 		if err != nil {
-			zap.S().Errorf("failed to verify authentication header: %v", err)
 			http.Error(w, fmt.Sprintf("Unauthorized: %v", err), http.StatusUnauthorized)
 			return
 		}

internal/api/syncapi/syncapi_test.go

@@ -539,6 +539,9 @@ func tryExpectExactOperations(t *testing.T, ctx context.Context, peer *peerUnder
 		for _, op := range ops {
 			op.Modno = 0
 		}
+		for _, op := range wantOps {
+			op.Modno = 0
+		}
 		if diff := cmp.Diff(ops, wantOps, protocmp.Transform()); diff != "" {
 			return fmt.Errorf("unexpected diff: %v", diff)
 		}

internal/api/syncapi/syncclient.go

@@ -182,11 +182,11 @@ func (c *SyncClient) RunSync(ctx context.Context) {
 		// Wait for the thread running the API loop and the thread running the stream connection to finish.
 		wg.Wait()
 
-		delay := c.reconnectDelay - time.Since(lastConnect)
+		reconnectDelayWithBackoff := c.reconnectDelay
 		if c.reconnectAttempts > 0 {
-			backoff := time.Duration(1<<min(c.reconnectAttempts, 5)) * c.reconnectDelay // 2^reconnectAttempts, max 32
-			delay += backoff
+			reconnectDelayWithBackoff *= time.Duration(1 << min(c.reconnectAttempts, 5)) // 2^reconnectAttempts, max 32
 		}
+		delay := reconnectDelayWithBackoff - time.Since(lastConnect)
 		c.l.Sugar().Infof("disconnected, will retry after %v (attempt %d)", delay, c.reconnectAttempts)
 		c.reconnectAttempts++
 		select {
@@ -362,7 +362,7 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
 	// This is a slow operation and we don't want to block the main loop waiting for it to complete and potentially forcing incomming messages to buffer or drop.
 	go func() {
 		startSync := func(diffSel *v1.OpSelector) error {
-			c.l.Sugar().Infof("starting sync with diffselector: %v", diffSel)
+			c.l.Sugar().Debugf("starting sync with diffselector: %v", diffSel)
 
 			diffQuery, err := protoutil.OpSelectorToQuery(diffSel)
 			if err != nil {
@@ -436,6 +436,11 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
 	return nil
 }
 
+func (c *syncSessionHandlerClient) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error {
+	c.l.Sugar().Infof("syncclient connection closed for client %q", c.peer.GetInstanceId())
+	return nil
+}
+
 func (c *syncSessionHandlerClient) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error {
 	c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(peerState *PeerState) {
 		if peerState == nil {

internal/api/syncapi/synccommon.go

@@ -25,16 +25,16 @@ func runSync(
 		return NewSyncErrorAuth(fmt.Errorf("peer not found in context, ensure authentication middleware is applied before sync handlers"))
 	}
 
-	if err := handler.OnConnectionEstablished(ctx, commandStream, peer); err != nil {
-		return err
-	}
-
 	defer func() {
 		if err := handler.OnConnectionClosed(ctx, commandStream); err != nil {
 			zap.L().Error("error handling connection closed", zap.Error(err))
 		}
 	}()
 
+	if err := handler.OnConnectionEstablished(ctx, commandStream, peer); err != nil {
+		return err
+	}
+
 	for item := range commandStream.ReadChannel() {
 		switch item.GetAction().(type) {
 		case *v1.SyncStreamItem_Heartbeat:
@@ -143,7 +143,7 @@ func (h *unimplementedSyncSessionHandler) OnConnectionEstablished(ctx context.Co
 }
 
 func (h *unimplementedSyncSessionHandler) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error {
-	return nil // no-op by default.
+	panic("must not be unimplemented")
 }
 
 func (h *unimplementedSyncSessionHandler) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error {

internal/api/syncapi/synchandler.go

@@ -220,13 +220,20 @@ func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context,
 	}()
 
 	// Send initial configuration to client
-	return h.sendConfigToClient(stream, h.snapshot.config)
+	if err := h.sendConfigToClient(stream, h.snapshot.config); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (h *syncSessionHandlerServer) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error {
-	h.mgr.mu.Lock()
-	delete(h.mgr.sessionHandlerMap, h.peer.Keyid)
-	h.mgr.mu.Unlock()
+	if h.peer != nil {
+		zap.S().Infof("syncserver connection closed for client %q", h.peer.InstanceId)
+		h.mgr.mu.Lock()
+		delete(h.mgr.sessionHandlerMap, h.peer.Keyid)
+		h.mgr.mu.Unlock()
+	}
 
 	// Close any active resources e.g. sinks for active log streams.
 	for logID, logSink := range h.activeLogStreams {
@@ -362,7 +369,7 @@ func (h *syncSessionHandlerServer) HandleDiffOperations(ctx context.Context, str
 func (h *syncSessionHandlerServer) HandleSendOperations(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendOperations) error {
 	switch event := item.GetEvent().Event.(type) {
 	case *v1.OperationEvent_CreatedOperations:
-		zap.L().Debug("syncserver received created operations", zap.Any("operations", event.CreatedOperations.GetOperations()))
+		zap.L().Debug("syncserver received create operations", zap.Any("operations", event.CreatedOperations.GetOperations()))
 		for _, op := range event.CreatedOperations.GetOperations() {
 			if err := h.insertOrUpdate(op); err != nil {
 				return fmt.Errorf("action SendOperations: operation event create %+v: %w", op, err)

internal/api/syncapi/syncmanager.go

@@ -151,6 +151,7 @@ func (m *SyncManager) RunSync(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
+			zap.S().Debugf("syncmanager context canceled for instance %q, stopping sync", m.snapshot.config.GetInstance())
 			return
 		case <-configWatchCh:
 			runSyncWithNewConfig()