fix: some fixes and new node test

examples/chat2-reliable/chat_reliability_test.go

@@ -524,3 +524,108 @@ func TestConflictResolution(t *testing.T) {
 	assert.Equal(t, env.chats[0].messageHistory[0].MessageId, env.chats[1].messageHistory[0].MessageId, "Conflicting messages should be ordered consistently")
 	assert.Equal(t, env.chats[0].messageHistory[1].MessageId, env.chats[1].messageHistory[1].MessageId, "Conflicting messages should be ordered consistently")
 }
+
+func TestNewNodeSyncAndMessagePropagation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	t.Log("Starting TestNewNodeSyncAndMessagePropagation")
+
+	// Set up initial network with 2 nodes
+	initialNodeCount := 2
+	env, err := setupTestEnvironment(ctx, t, initialNodeCount)
+	require.NoError(t, err, "Failed to set up initial test environment")
+
+	// Ensure initial nodes are connected
+	require.Eventually(t, func() bool {
+		return areNodesConnected(env.nodes, 1)
+	}, 60*time.Second, 1*time.Second, "Initial nodes failed to connect")
+
+	// Send some initial messages
+	t.Log("Sending initial messages")
+	env.chats[0].SendMessage("Initial message 1")
+	env.chats[1].SendMessage("Initial message 2")
+
+	// Wait for message propagation
+	time.Sleep(5 * time.Second)
+
+	// Verify initial messages are received by both nodes
+	for i, chat := range env.chats {
+		assert.Len(t, chat.messageHistory, 2, "Node %d should have 2 initial messages", i)
+	}
+
+	// Add a new node to the network
+	t.Log("Adding new node to the network")
+	newNode, err := setupTestNode(ctx, t)
+	require.NoError(t, err, "Failed to set up new node")
+	newChat, err := setupTestChat(ctx, newNode, "NewNode")
+	require.NoError(t, err, "Failed to set up new chat")
+
+	env.nodes = append(env.nodes, newNode)
+	env.chats = append(env.chats, newChat)
+
+	// Connect new node to the network
+	_, err = env.nodes[2].AddPeer(env.nodes[0].ListenAddresses()[0], peerstore.Static, env.chats[2].options.Relay.Topics.Value())
+	require.NoError(t, err, "Failed to connect new node to the network")
+
+	// Wait for the new node to sync
+	t.Log("Waiting for new node to sync")
+	// start := time.Now()
+	// syncTimeout := 2 * time.Minute
+	// require.Eventually(t, func() bool {
+	// 	msgCount := len(env.chats[2].messageHistory)
+	// 	t.Logf("New node message count: %d, Time elapsed: %v", msgCount, time.Since(start))
+	// 	return msgCount == 2
+	// }, syncTimeout, 5*time.Second, "New node failed to sync message history")
+	time.Sleep(60 * time.Second)
+
+	// Log the state of all nodes after sync
+	for i, chat := range env.chats {
+		t.Logf("Node %d message count: %d", i, len(chat.messageHistory))
+		for j, msg := range chat.messageHistory {
+			t.Logf("Node %d Message %d: %s", i, j, msg.Content)
+		}
+	}
+
+	// Send a message from an old node
+	t.Log("Sending message from old node")
+	env.chats[0].SendMessage("Message from old node")
+
+	// Wait for message propagation
+	time.Sleep(10 * time.Second)
+
+	// Verify the message is received by all nodes
+	for i, chat := range env.chats {
+		assert.Len(t, chat.messageHistory, 3, "Node %d should have 3 messages", i)
+	}
+
+	// Send a message from the new node
+	t.Log("Sending message from new node")
+	env.chats[2].SendMessage("Message from new node")
+
+	// Wait for message propagation
+	time.Sleep(10 * time.Second)
+
+	// Verify the message from the new node
+	assert.Len(t, env.chats[2].messageHistory, 4, "New node should have 4 messages (including its own)")
+
+	// Check if old nodes received the message from the new node
+	for i := 0; i < 2; i++ {
+		assert.Len(t, env.chats[i].messageHistory, 4, "Old node %d should have received the message from the new node", i)
+	}
+
+	for i := 0; i < 2; i++ {
+		lastMsg := env.chats[i].messageHistory[len(env.chats[i].messageHistory)-1]
+		assert.Equal(t, "Message from new node", lastMsg.Content, "Old node %d should have received the message from the new node", i)
+	}
+
+	// Log final state of all nodes
+	for i, chat := range env.chats {
+		t.Logf("Final state - Node %d message count: %d", i, len(chat.messageHistory))
+		for j, msg := range chat.messageHistory {
+			t.Logf("Node %d Message %d: %s", i, j, msg.Content)
+		}
+	}
+
+	t.Log("TestNewNodeSyncAndMessagePropagation completed")
+}

examples/chat2-reliable/go.mod

@@ -18,11 +18,14 @@ require (
 	github.com/ethereum/go-ethereum v1.10.26
 	github.com/google/uuid v1.4.0
 	github.com/ipfs/go-log/v2 v2.5.1
-	github.com/libp2p/go-libp2p v0.35.0
+	github.com/libp2p/go-libp2p v0.35.2
+	github.com/libp2p/go-msgio v0.3.0
 	github.com/muesli/reflow v0.3.0
 	github.com/multiformats/go-multiaddr v0.12.4
+	github.com/stretchr/testify v1.9.0
 	github.com/urfave/cli/v2 v2.27.2
 	github.com/waku-org/go-waku v0.2.3-0.20221109195301-b2a5a68d28ba
+	go.uber.org/zap v1.27.0
 	golang.org/x/term v0.20.0
 	google.golang.org/protobuf v1.34.1
 )
@@ -39,6 +42,7 @@ require (
 	github.com/btcsuite/btcd v0.20.1-beta // indirect
 	github.com/btcsuite/btcd/btcec/v2 v2.2.1 // indirect
 	github.com/btcsuite/btcutil v0.0.0-20190425235716-9e5f4b9a998d // indirect
+	github.com/cenkalti/backoff/v3 v3.2.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.1.2 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/containerd/cgroups v1.1.0 // indirect
@@ -62,7 +66,7 @@ require (
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/pprof v0.0.0-20240207164012-fb44976bdcd5 // indirect
-	github.com/gorilla/websocket v1.5.1 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/holiman/uint256 v1.2.2-0.20230321075855-87b91420868c // indirect
@@ -77,7 +81,6 @@ require (
 	github.com/libp2p/go-flow-metrics v0.1.0 // indirect
 	github.com/libp2p/go-libp2p-asn-util v0.4.1 // indirect
 	github.com/libp2p/go-libp2p-pubsub v0.11.0 // indirect
-	github.com/libp2p/go-msgio v0.3.0 // indirect
 	github.com/libp2p/go-nat v0.2.0 // indirect
 	github.com/libp2p/go-netroute v0.2.1 // indirect
 	github.com/libp2p/go-reuseport v0.4.0 // indirect
@@ -108,7 +111,7 @@ require (
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
 	github.com/pion/datachannel v1.5.6 // indirect
 	github.com/pion/dtls/v2 v2.2.11 // indirect
-	github.com/pion/ice/v2 v2.3.24 // indirect
+	github.com/pion/ice/v2 v2.3.25 // indirect
 	github.com/pion/interceptor v0.1.29 // indirect
 	github.com/pion/logging v0.2.2 // indirect
 	github.com/pion/mdns v0.0.12 // indirect
@@ -138,7 +141,6 @@ require (
 	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/status-im/status-go/extkeys v1.1.2 // indirect
-	github.com/stretchr/testify v1.9.0 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220614013038-64ee5596c38a // indirect
 	github.com/tklauser/go-sysconf v0.3.5 // indirect
 	github.com/tklauser/numcpus v0.2.2 // indirect
@@ -151,10 +153,9 @@ require (
 	github.com/wk8/go-ordered-map v1.0.0 // indirect
 	github.com/xrash/smetrics v0.0.0-20240312152122-5f08fbb34913 // indirect
 	go.uber.org/dig v1.17.1 // indirect
-	go.uber.org/fx v1.21.1 // indirect
+	go.uber.org/fx v1.22.1 // indirect
 	go.uber.org/mock v0.4.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.23.0 // indirect
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect
 	golang.org/x/mod v0.17.0 // indirect

examples/chat2-reliable/go.sum

@@ -100,6 +100,8 @@ github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792/go.mod h1:ghJtE
 github.com/btcsuite/winsvc v1.0.0/go.mod h1:jsenWakMcC0zFBFurPLEAyrnc/teJEM1O46fmI40EZs=
 github.com/buger/jsonparser v0.0.0-20181115193947-bf1c66bbce23/go.mod h1:bbYlZJ7hK1yFx9hf58LP0zeX7UjIGs20ufpu3evjr+s=
 github.com/c-bata/go-prompt v0.2.2/go.mod h1:VzqtzE2ksDBcdln8G7mk2RX9QyGjH+OVqOCSiVIqS34=
+github.com/cenkalti/backoff/v3 v3.2.2 h1:cfUAAO3yvKMYKPrvhDuHSwQnhZNk/RMHKdZqKTxfm6M=
+github.com/cenkalti/backoff/v3 v3.2.2/go.mod h1:cIeZDE3IrqwwJl6VUwCN6trj1oXrTS4rc0ij+ULvLYs=
 github.com/cenkalti/backoff/v4 v4.1.2 h1:6Yo7N8UP2K6LWZnW94DLVSSrbobcWdVzAYOisuDPIFo=
 github.com/cenkalti/backoff/v4 v4.1.2/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -300,8 +302,8 @@ github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5m
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
-github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/graph-gophers/graphql-go v1.3.0/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/grpc-gateway v1.5.0/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpgL2+G+NZTnrVHpWWfpdw=
@@ -400,8 +402,8 @@ github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6
 github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg=
 github.com/libp2p/go-flow-metrics v0.1.0 h1:0iPhMI8PskQwzh57jB9WxIuIOQ0r+15PChFGkx3Q3WM=
 github.com/libp2p/go-flow-metrics v0.1.0/go.mod h1:4Xi8MX8wj5aWNDAZttg6UPmc0ZrnFNsMtpsYUClFtro=
-github.com/libp2p/go-libp2p v0.35.0 h1:1xS1Bkr9X7GtdvV6ntLnDV9xB1kNjHK1lZ0eaO6gnhc=
-github.com/libp2p/go-libp2p v0.35.0/go.mod h1:snyJQix4ET6Tj+LeI0VPjjxTtdWpeOhYt5lEY0KirkQ=
+github.com/libp2p/go-libp2p v0.35.2 h1:287oHbuplkrLdAF+syB0n/qDgd50AUBtEODqS0e0HDs=
+github.com/libp2p/go-libp2p v0.35.2/go.mod h1:RKCDNt30IkFipGL0tl8wQW/3zVWEGFUZo8g2gAKxwjU=
 github.com/libp2p/go-libp2p-asn-util v0.4.1 h1:xqL7++IKD9TBFMgnLPZR6/6iYhawHKHl950SO9L6n94=
 github.com/libp2p/go-libp2p-asn-util v0.4.1/go.mod h1:d/NI6XZ9qxw67b4e+NgpQexCIiFYJjErASrYW4PFDN8=
 github.com/libp2p/go-libp2p-testing v0.12.0 h1:EPvBb4kKMWO29qP4mZGyhVzUyR25dvfUIK5WDu6iPUA=
@@ -557,8 +559,8 @@ github.com/pion/datachannel v1.5.6/go.mod h1:1eKT6Q85pRnr2mHiWHxJwO50SfZRtWHTsNI
 github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
 github.com/pion/dtls/v2 v2.2.11 h1:9U/dpCYl1ySttROPWJgqWKEylUdT0fXp/xst6JwY5Ks=
 github.com/pion/dtls/v2 v2.2.11/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
-github.com/pion/ice/v2 v2.3.24 h1:RYgzhH/u5lH0XO+ABatVKCtRd+4U1GEaCXSMjNr13tI=
-github.com/pion/ice/v2 v2.3.24/go.mod h1:KXJJcZK7E8WzrBEYnV4UtqEZsGeWfHxsNqhVcVvgjxw=
+github.com/pion/ice/v2 v2.3.25 h1:M5rJA07dqhi3nobJIg+uPtcVjFECTrhcR3n0ns8kDZs=
+github.com/pion/ice/v2 v2.3.25/go.mod h1:KXJJcZK7E8WzrBEYnV4UtqEZsGeWfHxsNqhVcVvgjxw=
 github.com/pion/interceptor v0.1.29 h1:39fsnlP1U8gw2JzOFWdfCU82vHvhW9o0rZnZF56wF+M=
 github.com/pion/interceptor v0.1.29/go.mod h1:ri+LGNjRUc5xUNtDEPzfdkmSqISixVTBF/z/Zms/6T4=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
@@ -772,8 +774,8 @@ go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/dig v1.17.1 h1:Tga8Lz8PcYNsWsyHMZ1Vm0OQOUaJNDyvPImgbAu9YSc=
 go.uber.org/dig v1.17.1/go.mod h1:Us0rSJiThwCv2GteUN0Q7OKvU7n5J4dxZ9JKUXozFdE=
-go.uber.org/fx v1.21.1 h1:RqBh3cYdzZS0uqwVeEjOX2p73dddLpym315myy/Bpb0=
-go.uber.org/fx v1.21.1/go.mod h1:HT2M7d7RHo+ebKGh9NRcrsrHHfpZ60nW3QRubMRfv48=
+go.uber.org/fx v1.22.1 h1:nvvln7mwyT5s1q201YE29V/BFrGor6vMiDNpU/78Mys=
+go.uber.org/fx v1.22.1/go.mod h1:HT2M7d7RHo+ebKGh9NRcrsrHHfpZ60nW3QRubMRfv48=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=

examples/chat2-reliable/peer_retrieval.go

@@ -23,16 +23,15 @@ import (
 const messageRequestProtocolID = protocol.ID("/chat2-reliable/message-request/1.0.0")
 
 // below functions are specifically for peer retrieval of missing msgs instead of store
-func (c *Chat) doRequestMissingMessageFromPeers(messageID string) error {
+func (c *Chat) doRequestMissingMessageFromPeers(messageID string) (*pb.Message, error) {
 	peers := c.node.Host().Network().Peers()
 	for _, peerID := range peers {
 		msg, err := c.requestMessageFromPeer(peerID, messageID)
 		if err == nil && msg != nil {
-			c.processReceivedMessage(msg)
-			return nil
+			return msg, nil
 		}
 	}
-	return errors.New("no peers could provide the missing message")
+	return nil, errors.New("no peers could provide the missing message")
 }
 
 func (c *Chat) requestMessageFromPeer(peerID peer.ID, messageID string) (*pb.Message, error) {
@@ -61,8 +60,6 @@ func (c *Chat) requestMessageFromPeer(peerID peer.ID, messageID string) (*pb.Mes
 		return nil, fmt.Errorf("failed to read message response: %w", err)
 	}
 
-	fmt.Printf("Received message response from peer %s\n", peerID.String())
-
 	if response.Message == nil {
 		return nil, fmt.Errorf("peer did not have the requested message")
 	}

examples/chat2-reliable/reliability.go

@@ -16,8 +16,8 @@ const (
 	bloomFilterWindow        = 1 * time.Hour
 	bloomFilterCleanInterval = 30 * time.Minute
 	bufferSweepInterval      = 5 * time.Second
-	syncMessageInterval      = 60 * time.Second
-	messageAckTimeout        = 10 * time.Second
+	syncMessageInterval      = 30 * time.Second
+	messageAckTimeout        = 60 * time.Second
 	maxRetries               = 3
 	retryBaseDelay           = 1 * time.Second
 	maxRetryDelay            = 10 * time.Second
@@ -88,7 +88,7 @@ func (c *Chat) SendMessage(line string) {
 		SenderId:         c.node.Host().ID().String(),
 		MessageId:        generateUniqueID(),
 		LamportTimestamp: c.getLamportTimestamp(),
-		CausalHistory:    c.getRecentMessageIDs(2),
+		CausalHistory:    c.getRecentMessageIDs(10),
 		ChannelId:        c.options.ContentTopic,
 		BloomFilter:      bloomBytes,
 		Content:          line,
@@ -111,25 +111,26 @@ func (c *Chat) SendMessage(line string) {
 		}
 		c.ui.ErrorMessage(err)
 	} else {
-		c.addToMessageHistory(msg)
 		c.bloomFilter.Add(msg.MessageId)
+		c.addToMessageHistory(msg)
+		c.ui.ChatMessage(int64(c.getLamportTimestamp()), msg.MessageId, msg.Content)
 	}
 }
 
 func (c *Chat) processReceivedMessage(msg *pb.Message) {
-	// Review ACK status of messages in the unacknowledged outgoing buffer
-	c.reviewAckStatus(msg)
-
 	// Check if the message is already in the bloom filter
 	if c.bloomFilter.Test(msg.MessageId) {
 		return
 	}
 
+	// Update bloom filter
+	c.bloomFilter.Add(msg.MessageId)
+
 	// Update Lamport timestamp
 	c.updateLamportTimestamp(msg.LamportTimestamp)
 
-	// Update bloom filter
-	c.bloomFilter.Add(msg.MessageId)
+	// Review ACK status of messages in the unacknowledged outgoing buffer
+	c.reviewAckStatus(msg)
 
 	// Check causal dependencies
 	missingDeps := c.checkCausalDependencies(msg)
@@ -156,18 +157,12 @@ func (c *Chat) processReceivedMessage(msg *pb.Message) {
 func (c *Chat) processBufferedMessages() {
 	c.mutex.Lock()
 
-	processed := make(map[string]bool)
 	remainingBuffer := make([]*pb.Message, 0, len(c.incomingBuffer))
 	processedBuffer := make([]*pb.Message, 0)
 
 	for _, msg := range c.incomingBuffer {
-		if processed[msg.MessageId] {
-			continue
-		}
-
 		missingDeps := c.checkCausalDependencies(msg)
 		if len(missingDeps) == 0 {
-			// Release the lock while processing the message
 			if msg.Content != "" {
 				c.ui.ChatMessage(int64(c.getLamportTimestamp()), msg.SenderId, msg.Content)
 			}
@@ -221,13 +216,12 @@ func (c *Chat) reviewAckStatus(msg *pb.Message) {
 
 func (c *Chat) requestMissingMessage(messageID string) {
 	for retry := 0; retry < maxRetries; retry++ {
-		err := c.doRequestMissingMessageFromPeers(messageID)
+		missedMsg, err := c.doRequestMissingMessageFromPeers(messageID)
 		if err == nil {
+			c.processReceivedMessage(missedMsg)
 			return
 		}
 
-		c.ui.ErrorMessage(fmt.Errorf("failed to retrieve missing message (attempt %d): %w", retry+1, err))
-
 		// Exponential backoff
 		delay := retryBaseDelay * time.Duration(1<<uint(retry))
 		if delay > maxRetryDelay {
@@ -236,7 +230,7 @@ func (c *Chat) requestMissingMessage(messageID string) {
 		time.Sleep(delay)
 	}
 
-	c.ui.ErrorMessage(fmt.Errorf("failed to retrieve missing message after %d attempts: %s", maxRetries, messageID))
+	c.ui.ErrorMessage(fmt.Errorf("failed to retrieve missing message %s after %d attempts", messageID, maxRetries))
 }
 
 func (c *Chat) checkCausalDependencies(msg *pb.Message) []string {
@@ -324,7 +318,7 @@ func (c *Chat) checkUnacknowledgedMessages() {
 				// Remove the message from the buffer after max attempts
 				c.outgoingBuffer = append(c.outgoingBuffer[:i], c.outgoingBuffer[i+1:]...)
 				i-- // Adjust index after removal
-				c.ui.ErrorMessage(fmt.Errorf("message %s dropped: failed to be acknowledged after %d attempts", unackMsg.Message.MessageId, maxResendAttempts))
+				c.ui.ErrorMessage(fmt.Errorf("message %s dropped: failed to be acknowledged after %d attempts", unackMsg.Message.Content, maxResendAttempts))
 			}
 		}
 	}
@@ -412,9 +406,7 @@ func (c *Chat) updateLamportTimestamp(msgTs int32) {
 	c.lamportTSMutex.Lock()
 	defer c.lamportTSMutex.Unlock()
 	if msgTs > c.lamportTimestamp {
-		c.lamportTimestamp = msgTs + 1
-	} else {
-		c.lamportTimestamp++
+		c.lamportTimestamp = msgTs
 	}
 }