update

pkg/network/anonymity/anonymity.go

@@ -77,9 +77,7 @@ func (p *sNode) Run(pCtx context.Context) error {
 			p.fNetwork.HandleFunc(p.fSettings.GetNetworkMask(), nil)
 			return nil
 		}
-		if err := p.fState.Disable(disableFunc); err != nil {
-			panic(err)
-		}
+		_ = p.fState.Disable(disableFunc)
 	}()
 
 	chErr := make(chan error)
@@ -174,15 +172,23 @@ func (p *sNode) FetchPayload(pCtx context.Context, pRecv asymmetric.IPubKey, pPl
 }
 
 func (p *sNode) enqueueMessage(pCtx context.Context, pMsg message.IMessage) error {
-	for i := uint64(0); i <= p.fSettings.GetRetryEnqueue(); i++ {
-		if err := p.fQueue.EnqueueMessage(pMsg); err == nil {
-			return nil
-		}
+	retryNum := p.fSettings.GetRetryEnqueue()
+	for i := uint64(0); i <= retryNum; i++ {
 		select {
 		case <-pCtx.Done():
 			return pCtx.Err()
-		case <-time.After(p.fQueue.GetSettings().GetDuration()):
-			// next iter
+		default:
+			if err := p.fQueue.EnqueueMessage(pMsg); err == nil {
+				return nil
+			}
+			if i != retryNum {
+				select {
+				case <-pCtx.Done():
+					return pCtx.Err()
+				case <-time.After(p.fQueue.GetSettings().GetDuration()):
+					break // in the next iteration, a repetition occurs
+				}
+			}
 		}
 	}
 	return ErrEnqueueMessage

pkg/network/anonymity/anonymity_test.go

@@ -37,7 +37,7 @@ const (
 func TestNodeSettings(t *testing.T) {
 	t.Parallel()
 
-	node, cancels := testNewNode(time.Minute, "", 9, 0, false)
+	node, cancels := testNewNode(time.Minute, "", 9, 0, 0, false)
 	defer testFreeNodes([]INode{node}, []context.CancelFunc{cancels}, 9)
 
 	sett := node.GetSettings()
@@ -355,7 +355,7 @@ func TestEnqueuePayload(t *testing.T) {
 func TestHandleWrapper(t *testing.T) {
 	t.Parallel()
 
-	_node, cancel := testNewNode(time.Minute, "", 7, 0, true)
+	_node, cancel := testNewNode(time.Minute, "", 7, 0, 0, true)
 	defer testFreeNodes([]INode{_node}, []context.CancelFunc{cancel}, 7)
 
 	node := _node.(*sNode)
@@ -471,7 +471,7 @@ func TestHandleWrapper(t *testing.T) {
 func TestStoreHashWithBroadcastMessage(t *testing.T) {
 	t.Parallel()
 
-	_node, cancel := testNewNode(time.Minute, "", 6, 0, false)
+	_node, cancel := testNewNode(time.Minute, "", 6, 0, 0, false)
 	defer testFreeNodes([]INode{_node}, []context.CancelFunc{cancel}, 6)
 
 	node := _node.(*sNode)
@@ -520,7 +520,7 @@ func TestStoreHashWithBroadcastMessage(t *testing.T) {
 func TestRecvSendMessage(t *testing.T) {
 	t.Parallel()
 
-	_node, cancel := testNewNode(time.Minute, "", 5, 0, false)
+	_node, cancel := testNewNode(time.Minute, "", 5, 0, 0, false)
 	defer testFreeNodes([]INode{_node}, []context.CancelFunc{cancel}, 5)
 
 	node := _node.(*sNode)
@@ -569,14 +569,56 @@ func TestRecvSendMessage(t *testing.T) {
 		}
 	}
 
+	hasError := false
 	for i := 0; i < 10; i++ {
 		// message can be dequeued in the send's call time
 		if err := node.enqueueMessage(ctx, msg); err != nil {
-			return
+			hasError = true
+			break
 		}
 	}
 
-	t.Error("success send message (push to queue) over queue capacity")
+	if !hasError {
+		t.Error("success send message (push to queue) over queue capacity")
+	}
+}
+
+func TestRetryEnqueue(t *testing.T) {
+	t.Parallel()
+
+	ctxBg, cancelBg := context.WithCancel(context.Background())
+	defer cancelBg()
+
+	_node, cancel := testNewNode(time.Minute, "", 11, 0, 3, false)
+	defer testFreeNodes([]INode{_node}, []context.CancelFunc{cancel}, 11)
+
+	node := _node.(*sNode)
+
+	client := node.fQueue.GetClient()
+	pubKey := client.GetPubKey()
+
+	msgBody := "hello, world!"
+	msg, err := client.EncryptPayload(
+		pubKey,
+		adapters.NewPayload(
+			testutils.TcHead,
+			wrapRequest([]byte(msgBody)),
+		).ToOrigin(),
+	)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	go func() {
+		for i := 0; i < testutils.TCQueueCapacity*2; i++ {
+			_ = node.enqueueMessage(ctxBg, msg)
+		}
+	}()
+
+	time.Sleep(2 * time.Second)
+	cancelBg()
+	time.Sleep(time.Second)
 }
 
 // nodes[0], nodes[1] = clients
@@ -589,7 +631,7 @@ func testNewNodes(t *testing.T, timeWait time.Duration, addresses [2]string, typ
 	addrs := [5]string{"", "", addresses[0], "", addresses[1]}
 
 	for i := 0; i < 5; i++ {
-		nodes[i], cancels[i] = testNewNode(timeWait, addrs[i], typeDB, i, false)
+		nodes[i], cancels[i] = testNewNode(timeWait, addrs[i], typeDB, i, 0, false)
 		if nodes[i] == nil {
 			t.Errorf("node (%d) is not running %d", i, typeDB)
 			return [5]INode{}, [5]context.CancelFunc{}
@@ -658,7 +700,7 @@ func testNewNodes(t *testing.T, timeWait time.Duration, addresses [2]string, typ
 	return nodes, cancels
 }
 
-func testNewNode(timeWait time.Duration, addr string, typeDB, numDB int, f2fDisabled bool) (INode, context.CancelFunc) {
+func testNewNode(timeWait time.Duration, addr string, typeDB, numDB, retryNum int, f2fDisabled bool) (INode, context.CancelFunc) {
 	db, err := database.NewKVDatabase(
 		storage.NewSettings(&storage.SSettings{
 			FPath:     fmt.Sprintf(tcPathDBTemplate, typeDB, numDB),
@@ -677,6 +719,7 @@ func testNewNode(timeWait time.Duration, addr string, typeDB, numDB int, f2fDisa
 			FF2FDisabled:   f2fDisabled,
 			FNetworkMask:   networkMask,
 			FFetchTimeWait: timeWait,
+			FRetryEnqueue:  uint64(retryNum),
 		}),
 		logger.NewLogger(
 			logger.NewSettings(&logger.SSettings{}),

pkg/network/anonymity/queue/queue.go

@@ -62,11 +62,7 @@ func (p *sMessageQueue) Run(pCtx context.Context) error {
 	if err := p.fState.Enable(nil); err != nil {
 		return utils.MergeErrors(ErrRunning, err)
 	}
-	defer func() {
-		if err := p.fState.Disable(nil); err != nil {
-			panic(err)
-		}
-	}()
+	defer func() { _ = p.fState.Disable(nil) }()
 
 	for {
 		select {

pkg/network/anonymity/queue/queue_test.go

@@ -162,7 +162,10 @@ func TestQueue(t *testing.T) {
 
 func testQueue(queue IMessageQueue) error {
 	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	defer func() {
+		cancel()
+		time.Sleep(100 * time.Millisecond)
+	}()
 
 	go func() {
 		if err := queue.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
@@ -170,6 +173,19 @@ func testQueue(queue IMessageQueue) error {
 		}
 	}()
 
+	client := queue.GetClient()
+	msg, err := client.EncryptPayload(
+		client.GetPubKey(),
+		payload.NewPayload(0, []byte(testutils.TcBody)),
+	)
+	if err != nil {
+		return err
+	}
+
+	if err := queue.EnqueueMessage(msg); err != nil {
+		return err
+	}
+
 	// wait minimum one generated message
 	time.Sleep(200 * time.Millisecond)
 
@@ -199,8 +215,7 @@ func testQueue(queue IMessageQueue) error {
 		}
 	}
 
-	client := queue.GetClient()
-	msg, err := client.EncryptPayload(
+	msg2, err := client.EncryptPayload(
 		client.GetPubKey(),
 		payload.NewPayload(0, []byte(testutils.TcBody)),
 	)
@@ -208,9 +223,11 @@ func testQueue(queue IMessageQueue) error {
 		return err
 	}
 
-	hash := msg.GetHash()
+	hash := msg2.GetHash()
 	for i := 0; i < 3; i++ {
-		queue.EnqueueMessage(msg)
+		if err := queue.EnqueueMessage(msg2); err != nil {
+			return err
+		}
 	}
 
 	for i := 0; i < 3; i++ {

pkg/network/conn/conn_test.go

@@ -110,7 +110,9 @@ func TestClosedConn(t *testing.T) {
 	pld := payload.NewPayload(1, []byte("aaa"))
 	msg := message.NewMessage(conn.GetSettings(), pld, 1)
 
-	ctx := context.Background()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	if err := conn.WriteMessage(ctx, msg); err == nil {
 		t.Error("success write payload to closed connection")
 		return
@@ -142,6 +144,18 @@ func TestClosedConn(t *testing.T) {
 		t.Error("success recv head bytes from closed connection")
 		return
 	}
+
+	cancel()
+
+	if err := sconn.sendBytes(ctx, []byte("hello, world!")); err == nil {
+		t.Error("success send bytes with canceled context")
+		return
+	}
+
+	if _, err := sconn.recvDataBytes(ctx, 128, time.Second); err == nil {
+		t.Error("success recv data bytes with canceled context")
+		return
+	}
 }
 
 func TestInvalidConn(t *testing.T) {

pkg/network/network.go

@@ -91,6 +91,8 @@ func (p *sNode) BroadcastMessage(pCtx context.Context, pMsg message.IMessage) er
 			}()
 
 			select {
+			case <-pCtx.Done():
+				resErr = pCtx.Err()
 			case err := <-chErr:
 				resErr = err // err can be = nil
 			case <-time.After(p.fSettings.GetWriteTimeout()):

pkg/network/network_test.go

@@ -315,6 +315,44 @@ func TestNodeSettings(t *testing.T) {
 	}
 }
 
+func TestContextCancel(t *testing.T) {
+	t.Parallel()
+
+	node1 := newTestNode(testutils.TgAddrs[60], testutils.TCMaxConnects, time.Minute)
+	node2 := newTestNode("", testutils.TCMaxConnects, time.Minute)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go func() { _ = node1.Listen(ctx) }()
+
+	err1 := testutils.TryN(50, 10*time.Millisecond, func() error {
+		return node2.AddConnection(ctx, testutils.TgAddrs[60])
+	})
+	if err1 != nil {
+		t.Error(err1)
+		return
+	}
+
+	headHandle := uint64(testutils.TcHead)
+	sett := node2.GetSettings().GetConnSettings()
+
+	go func() {
+		for i := 0; i < 1000; i++ {
+			pld := payload.NewPayload(
+				headHandle,
+				[]byte(fmt.Sprintf(testutils.TcBodyTemplate, i)),
+			)
+			if err := node2.BroadcastMessage(ctx, message.NewMessage(sett, pld, 1)); err != nil {
+				return
+			}
+		}
+		t.Error("success all broadcast messages with canceled context")
+	}()
+
+	cancel()
+}
+
 func testNodes() ([5]INode, map[INode]map[string]bool, error) {
 	nodes := [5]INode{}
 	addrs := [5]string{"", "", testutils.TgAddrs[0], "", testutils.TgAddrs[1]}

test/result/badge.svg

@@ -1 +1 @@
-<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="96" height="20" role="img" aria-label="coverage: 96%"><title>coverage: 96%</title><linearGradient id="s" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="r"><rect width="96" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#r)"><rect width="61" height="20" fill="#555"/><rect x="61" width="35" height="20" fill="#97ca00"/><rect width="96" height="20" fill="url(#s)"/></g><g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" text-rendering="geometricPrecision" font-size="110"><text aria-hidden="true" x="315" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="510">coverage</text><text x="315" y="140" transform="scale(.1)" fill="#fff" textLength="510">coverage</text><text aria-hidden="true" x="775" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="250">96%</text><text x="775" y="140" transform="scale(.1)" fill="#fff" textLength="250">96%</text></g></svg>
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="96" height="20" role="img" aria-label="coverage: 97%"><title>coverage: 97%</title><linearGradient id="s" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="r"><rect width="96" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#r)"><rect width="61" height="20" fill="#555"/><rect x="61" width="35" height="20" fill="#97ca00"/><rect width="96" height="20" fill="url(#s)"/></g><g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" text-rendering="geometricPrecision" font-size="110"><text aria-hidden="true" x="315" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="510">coverage</text><text x="315" y="140" transform="scale(.1)" fill="#fff" textLength="510">coverage</text><text aria-hidden="true" x="775" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="250">97%</text><text x="775" y="140" transform="scale(.1)" fill="#fff" textLength="250">97%</text></g></svg>

test/utils/testutils.go

@@ -37,7 +37,29 @@ var (
 )
 
 var (
-	TgAddrs = [60]string{
+	TgAddrs = [80]string{
+		"localhost:8020",
+		"localhost:8021",
+		"localhost:8022",
+		"localhost:8023",
+		"localhost:8024",
+		"localhost:8025",
+		"localhost:8026",
+		"localhost:8027",
+		"localhost:8028",
+		"localhost:8029",
+
+		"localhost:8030",
+		"localhost:8031",
+		"localhost:8032",
+		"localhost:8033",
+		"localhost:8034",
+		"localhost:8035",
+		"localhost:8036",
+		"localhost:8037",
+		"localhost:8038",
+		"localhost:8039",
+
 		"localhost:8040",
 		"localhost:8041",
 		"localhost:8042",