Fixed find last transaction for delayed msg in & Fixed dht findValue …

README.md

@@ -4,7 +4,7 @@
 
 [![Based on TON][ton-svg]][ton]
 [![Telegram Channel][tgc-svg]][tg-channel]
-![Coverage](https://img.shields.io/badge/Coverage-70.4%25-brightgreen)
+![Coverage](https://img.shields.io/badge/Coverage-70.5%25-brightgreen)
 
 Golang library for interacting with TON blockchain.
 

adnl/dht/client.go

@@ -450,6 +450,7 @@ func (c *Client) FindValue(ctx context.Context, key *Key, continuation ...*Conti
 
 	cond := sync.NewCond(&sync.Mutex{})
 	waitingThreads := 0
+	stopped := false
 
 	launchWorker := func() {
 		for {
@@ -465,12 +466,18 @@ func (c *Client) FindValue(ctx context.Context, key *Key, continuation ...*Conti
 			for node == nil {
 				waitingThreads++
 				if waitingThreads == threads {
+					stopped = true
+					cond.Broadcast()
 					cond.L.Unlock()
 					result <- nil
 					return
 				}
 
 				cond.Wait()
+				if stopped {
+					cond.L.Unlock()
+					return
+				}
 				node, _ = plist.Get()
 				waitingThreads--
 			}
@@ -485,20 +492,27 @@ func (c *Client) FindValue(ctx context.Context, key *Key, continuation ...*Conti
 
 			switch v := val.(type) {
 			case *Value:
+				cond.L.Lock()
+				if !stopped {
+					stopped = true
+					cond.Broadcast()
+				}
+				cond.L.Unlock()
 				result <- &foundResult{value: v, node: node}
 				return
 			case []*Node:
 				added := false
+				cond.L.Lock()
 				for _, n := range v {
 					if newNode, err := c.addNode(n); err == nil {
 						plist.Add(newNode)
 						added = true
 					}
 				}
-
 				if added {
 					cond.Broadcast()
 				}
+				cond.L.Unlock()
 			}
 		}
 	}
@@ -509,8 +523,20 @@ func (c *Client) FindValue(ctx context.Context, key *Key, continuation ...*Conti
 
 	select {
 	case <-ctx.Done():
+		cond.L.Lock()
+		if !stopped {
+			stopped = true
+			cond.Broadcast()
+		}
+		cond.L.Unlock()
 		return nil, nil, ctx.Err()
 	case val := <-result:
+		cond.L.Lock()
+		if !stopped {
+			stopped = true
+			cond.Broadcast()
+		}
+		cond.L.Unlock()
 		if val == nil {
 			return nil, cont, ErrDHTValueIsNotFound
 		}

adnl/rldp/bbr2.go

@@ -153,7 +153,7 @@ func NewBBRv2Controller(l *TokenBucket, o BBRv2Options) *BBRv2Controller {
 
 	start := l.GetRate()
 	if start <= 0 {
-		start = max64(o.MinRate, 1024*64) // 64KiB/s как нижний разумный
+		start = max64(o.MinRate, 1024*64)
 	}
 	c.btlbw.Store(start)
 	c.pacingRate.Store(start)

adnl/rldp/client.go

@@ -70,10 +70,12 @@ type activeTransferPart struct {
 type activeTransfer struct {
 	id        []byte
 	data      []byte
+	totalSize uint64
 	timeoutAt int64
 
-	currentPart atomic.Pointer[activeTransferPart]
-	rldp        *RLDP
+	nextPartIndex uint32
+	currentPart   atomic.Pointer[activeTransferPart]
+	rldp          *RLDP
 
 	mx sync.Mutex
 }
@@ -840,7 +842,7 @@ func (r *RLDP) recoverySender() {
 							TransferID: part.transfer.id,
 							FecType:    part.fec,
 							Part:       part.index,
-							TotalSize:  uint64(len(part.transfer.data)),
+							TotalSize:  part.transfer.totalSize,
 							Seqno:      seqno,
 							Data:       part.encoder.GenSymbol(seqno),
 						}
@@ -924,9 +926,24 @@ func (r *RLDP) recoverySender() {
 				r.rateCtrl.SetAppLimited(false)
 			}
 
+			for i := range transfersToProcess {
+				transfersToProcess[i] = nil
+			}
 			transfersToProcess = transfersToProcess[:0]
+
+			for i := range timedOut {
+				timedOut[i] = nil
+			}
 			timedOut = timedOut[:0]
+
+			for i := range timedOutReq {
+				timedOutReq[i] = ""
+			}
 			timedOutReq = timedOutReq[:0]
+
+			for i := range timedOutExp {
+				timedOutExp[i] = ""
+			}
 			timedOutExp = timedOutExp[:0]
 		}
 	}
@@ -937,6 +954,7 @@ func (r *RLDP) startTransfer(ctx context.Context, transferId, data []byte, recov
 		id:        transferId,
 		timeoutAt: recoverTimeoutAt * 1000, // ms
 		data:      data,
+		totalSize: uint64(len(data)),
 		rldp:      r,
 	}
 
@@ -974,21 +992,22 @@ func (t *activeTransfer) prepareNextPart() (bool, error) {
 		return false, nil // fmt.Errorf("transfer timed out")
 	}
 
-	partIndex := uint32(0)
-	if cp := t.getCurrentPart(); cp != nil {
-		partIndex = cp.index + 1
-	}
-
-	if len(t.data) <= int(partIndex*PartSize) {
-		// all parts sent
+	if len(t.data) == 0 {
 		return false, nil
 	}
 
-	payload := t.data[partIndex*PartSize:]
+	partIndex := t.nextPartIndex
+
+	payload := t.data
 	if len(payload) > int(PartSize) {
 		payload = payload[:PartSize]
 	}
 
+	if len(payload) == 0 {
+		return false, nil
+	}
+	remaining := t.data[len(payload):]
+
 	cnt := uint32(len(payload))/DefaultSymbolSize + 1
 
 	var err error
@@ -1029,10 +1048,24 @@ func (t *activeTransfer) prepareNextPart() (bool, error) {
 		fecSymbolSize:    fec.GetSymbolSize(),
 		nextRecoverDelay: 15,
 		fastSeqnoTill:    cnt + cnt/50 + 1, // +2%
-		sendClock:        NewSendClock(32 << 10),
 		transfer:         t,
 	}
 
+	pt := uint32(1) << uint32(math.Ceil(math.Log2(float64(part.fecSymbolsCount))))
+	if pt > 16<<10 {
+		pt = 16 << 10
+	} else if pt < 64 {
+		pt = 64
+	}
+	part.sendClock = NewSendClock(int(pt))
+
+	if len(remaining) == 0 {
+		t.data = nil
+	} else {
+		t.data = remaining
+	}
+
+	t.nextPartIndex++
 	t.currentPart.Store(&part)
 	return true, nil
 }
@@ -1047,7 +1080,7 @@ func (r *RLDP) sendFastSymbols(ctx context.Context, transfer *activeTransfer) er
 		TransferID: transfer.id,
 		FecType:    part.fec,
 		Part:       part.index,
-		TotalSize:  uint64(len(transfer.data)),
+		TotalSize:  transfer.totalSize,
 	}
 
 	sc := part.fastSeqnoTill
@@ -1124,22 +1157,28 @@ type AsyncQueryResult struct {
 }
 
 func (r *RLDP) DoQueryAsync(ctx context.Context, maxAnswerSize uint64, id []byte, query tl.Serializable, result chan<- AsyncQueryResult) error {
-	timeout, ok := ctx.Deadline()
-	if !ok {
-		timeout = time.Now().Add(15 * time.Second)
-	}
-
 	if len(id) != 32 {
 		return errors.New("invalid id")
 	}
 
+	now := time.Now()
+	timeout, ok := ctx.Deadline()
+	if !ok {
+		timeout = now.Add(15 * time.Second)
+	}
+
 	q := &Query{
 		ID:            id,
 		MaxAnswerSize: maxAnswerSize,
 		Timeout:       uint32(timeout.Unix()),
 		Data:          query,
 	}
 
+	if uxMin := now.Unix() + 2; int64(q.Timeout) < uxMin {
+		// because timeout in seconds, we should add some to avoid an early drop
+		q.Timeout = uint32(uxMin)
+	}
+
 	data, err := tl.Serialize(q, true)
 	if err != nil {
 		return fmt.Errorf("failed to serialize query: %w", err)
@@ -1197,6 +1236,11 @@ func (r *RLDP) SendAnswer(ctx context.Context, maxAnswerSize uint64, timeoutAt u
 		tm = int64(timeoutAt)
 	}
 
+	if minT := time.Now().Unix() + 1; tm < minT {
+		// give at least 1 sec in case of a clock problem
+		tm = minT
+	}
+
 	if err = r.startTransfer(ctx, reverseTransferId(toTransferId), data, tm); err != nil {
 		return fmt.Errorf("failed to send partitioned answer: %w", err)
 	}

adnl/rldp/queue_test.go

@@ -28,7 +28,7 @@ func TestQueue_Basic(t *testing.T) {
 		if !ok {
 			t.Fatalf("expected ok for i=%d", i)
 		}
-		
+
 		if m.Seqno != i {
 			t.Fatalf("want=%d got=%d", i, m.Seqno)
 		}
@@ -43,7 +43,6 @@ func TestQueue_Basic(t *testing.T) {
 func TestQueue_OverwriteOldest(t *testing.T) {
 	q := NewQueue(4)
 
-	// кладём 6 без чтения — должны остаться последние 4: 2,3,4,5
 	for i := uint32(0); i < 6; i++ {
 		q.Enqueue(mp(i))
 	}
@@ -67,11 +66,10 @@ func TestQueue_OverwriteOldest(t *testing.T) {
 func TestQueue_OverwriteInterleaved(t *testing.T) {
 	q := NewQueue(2)
 
-	// вместимость 2: заполним и начнем выталкивать
 	q.Enqueue(mp(0))
 	q.Enqueue(mp(1))
-	q.Enqueue(mp(2)) // вытолкнет 0
-	q.Enqueue(mp(3)) // вытолкнет 1
+	q.Enqueue(mp(2))
+	q.Enqueue(mp(3))
 
 	// ожидаем 2,3
 	m, ok := q.Dequeue()

go.mod

@@ -6,6 +6,6 @@ toolchain go1.24.3
 
 require (
 	filippo.io/edwards25519 v1.1.0
-	github.com/xssnick/raptorq v1.2.0
+	github.com/xssnick/raptorq v1.3.0
 	golang.org/x/crypto v0.42.0
 )

go.sum

@@ -1,6 +1,6 @@
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
-github.com/xssnick/raptorq v1.2.0 h1:ts8yjB3Xns3GS4V7/xQLx9lYLAlSwzlvJjGDLscZCDA=
-github.com/xssnick/raptorq v1.2.0/go.mod h1:kgEVVsZv2hP+IeV7C7985KIFsDdvYq2ARW234SBA9Q4=
+github.com/xssnick/raptorq v1.3.0 h1:3GoaySKMg/i8rbjhIuqjxpTTO2l3Gs2/Gh7k3GAjvGo=
+github.com/xssnick/raptorq v1.3.0/go.mod h1:kgEVVsZv2hP+IeV7C7985KIFsDdvYq2ARW234SBA9Q4=
 golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI=
 golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8=

ton/transactions.go

@@ -362,8 +362,6 @@ func (c *APIClient) findLastTransactionByHash(ctx context.Context, addr *address
 						return transaction, nil
 					}
 				}
-
-				continue
 			} else {
 				if transaction.IO.In == nil {
 					continue
@@ -378,8 +376,6 @@ func (c *APIClient) findLastTransactionByHash(ctx context.Context, addr *address
 					return transaction, nil
 				}
 			}
-
-			return transaction, nil
 		}
 
 		scanned += 15