Update counting msg size

commonspace/object/tree/synctree/headupdate.go

@@ -15,9 +15,13 @@ type InnerHeadUpdate struct {
 	root         *treechangeproto.RawTreeChangeWithId
 }
 
-func (h InnerHeadUpdate) MsgSize() uint64 {
-	size := uint64(len(h.heads))
-	size += uint64(len(h.snapshotPath))
+func (h InnerHeadUpdate) MsgSize() (size uint64) {
+	for _, head := range h.heads {
+		size += uint64(len(head))
+	}
+	for _, snapshotId := range h.snapshotPath {
+		size += uint64(len(snapshotId))
+	}
 	for _, change := range h.changes {
 		size += uint64(len(change.Id))
 		size += uint64(len(change.RawChange))

commonspace/sync/syncmetric.go

@@ -73,8 +73,15 @@ func (m *syncMetric) UpdateQueueSize(size uint64, msgType int, add bool) {
 		atCount.Add(1)
 		m.totalSize.Add(intSize)
 	} else {
-		atSize.Add(-intSize)
-		atCount.Add(-1)
-		m.totalSize.Add(-intSize)
+		curCount := atCount.Load()
+		curSize := atSize.Load()
+		if curCount != 0 {
+			atCount.Add(-1)
+		}
+		// TODO: fix the root cause :-)
+		if curSize > intSize {
+			atSize.Add(-intSize)
+			m.totalSize.Add(-intSize)
+		}
 	}
 }

util/multiqueue/multiqueue.go

@@ -71,11 +71,12 @@ func (m *multiQueue[T]) Add(ctx context.Context, threadId string, msg T) (err er
 		q = m.startThread(threadId)
 	}
 	m.mu.Unlock()
+	m.updateSize(msg, true)
 	err = q.TryAdd(msg)
 	if err != nil {
+		m.updateSize(msg, false)
 		return
 	}
-	m.updateSize(msg, true)
 	return
 }