[Server] Move db write ops to background (#512)

* Move db write ops to background

* Impl retry logic at repo level

* Drop retry logic from app layer

* Fix

* Fix tests

* Lint

* Fix wasm wrapper

Author: altafan

pkg/client-sdk/test/wasm/wasm_test.go

@@ -121,14 +121,20 @@ func TestWasm(t *testing.T) {
 	require.NoError(t, err)
 
 	time.Sleep(3 * time.Second)
-	err = generateBlock()
-	require.NoError(t, err)
-	time.Sleep(2 * time.Second)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		waitForPayment(alicePage, aliceAddr.OffchainAddr)
+	}()
 
 	txID, err := settle(alicePage)
 	require.NoError(t, err)
 	t.Logf("Alice onboard txID: %v", txID)
 
+	wg.Wait()
+
 	aliceBalance, err := getBalance(alicePage)
 	require.NoError(t, err)
 	t.Logf("Alice onchain balance: %v", aliceBalance.OnchainSpendable)
@@ -141,20 +147,32 @@ func TestWasm(t *testing.T) {
 
 	amount := 1000
 	t.Logf("Alice is sending %d sats to Bob offchain...", amount)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		waitForPayment(alicePage, bobAddr.OffchainAddr)
+	}()
 	require.NoError(t, sendOffChain(alicePage, bobAddr.OffchainAddr, amount))
 
 	t.Log("Transaction completed out of round")
 
+	wg.Wait()
+
 	t.Logf("Bob settling the received funds...")
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		waitForPayment(bobPage, bobAddr.OffchainAddr)
+	}()
 	txID, err = settle(bobPage)
 	require.NoError(t, err)
 	t.Logf("Bob settled the received funds in round %v", txID)
 
+	wg.Wait()
+
 	err = generateBlock()
 	require.NoError(t, err)
 
-	time.Sleep(5 * time.Second)
-
 	aliceBalance, err = getBalance(alicePage)
 	require.NoError(t, err)
 	t.Logf("Alice onchain balance: %v", aliceBalance.OnchainSpendable)
@@ -350,6 +368,18 @@ func sendOffChain(page playwright.Page, addr string, amount int) error {
 	return err
 }
 
+func waitForPayment(page playwright.Page, addr string) error {
+	_, err := page.Evaluate(fmt.Sprintf(`async () => {
+			try {
+            return await notifyIncomingFunds("%s");
+        } catch (err) {
+            console.error("Error:", err);
+            throw err;
+        }
+	}`, addr))
+	return err
+}
+
 func runCommand(name string, arg ...string) (string, error) {
 	errb := new(strings.Builder)
 	cmd := newCommand(name, arg...)

pkg/client-sdk/wasm/browser/wrappers.go

@@ -574,7 +574,7 @@ func NotifyIncomingFundsWrapper() js.Func {
 			return nil, err
 		}
 
-		return result, nil
+		return js.ValueOf(string(result)), nil
 	})
 }
 

server/internal/core/application/covenantless.go

@@ -524,54 +524,53 @@ func (s *covenantlessService) SubmitRedeemTx(
 		return "", "", fmt.Errorf("failed to sign redeem tx: %s", err)
 	}
 
-	// Create new vtxos, update spent vtxos state
-	newVtxos := make([]domain.Vtxo, 0, len(ptx.UnsignedTx.TxOut))
-	for outIndex, out := range outputs {
-		vtxoTapKey, err := schnorr.ParsePubKey(out.PkScript[2:])
-		if err != nil {
-			return "", "", fmt.Errorf("failed to parse vtxo taproot key: %s", err)
+	go func(ptx *psbt.Packet, signedRedeemTx, redeemTxid string) {
+		ctx := context.Background()
+		// Create new vtxos, update spent vtxos state
+		newVtxos := make([]domain.Vtxo, 0, len(ptx.UnsignedTx.TxOut))
+		for outIndex, out := range outputs {
+			//notlint:all
+			vtxoPubkey := hex.EncodeToString(out.PkScript[2:])
+
+			newVtxos = append(newVtxos, domain.Vtxo{
+				VtxoKey: domain.VtxoKey{
+					Txid: redeemTxid,
+					VOut: uint32(outIndex),
+				},
+				PubKey:    vtxoPubkey,
+				Amount:    uint64(out.Value),
+				ExpireAt:  expiration,
+				RoundTxid: roundTxid,
+				RedeemTx:  signedRedeemTx,
+				CreatedAt: time.Now().Unix(),
+			})
 		}
 
-		vtxoPubkey := hex.EncodeToString(schnorr.SerializePubKey(vtxoTapKey))
-
-		newVtxos = append(newVtxos, domain.Vtxo{
-			VtxoKey: domain.VtxoKey{
-				Txid: redeemTxid,
-				VOut: uint32(outIndex),
-			},
-			PubKey:    vtxoPubkey,
-			Amount:    uint64(out.Value),
-			ExpireAt:  expiration,
-			RoundTxid: roundTxid,
-			RedeemTx:  signedRedeemTx,
-			CreatedAt: time.Now().Unix(),
-		})
-	}
+		if err := s.repoManager.Vtxos().AddVtxos(ctx, newVtxos); err != nil {
+			log.WithError(err).Warn("failed to add vtxos")
+			return
+		}
+		log.Debugf("added %d vtxos", len(newVtxos))
 
-	if err := s.repoManager.Vtxos().AddVtxos(ctx, newVtxos); err != nil {
-		return "", "", fmt.Errorf("failed to add vtxos: %s", err)
-	}
-	log.Infof("added %d vtxos", len(newVtxos))
-	if err := s.startWatchingVtxos(newVtxos); err != nil {
-		log.WithError(err).Warn(
-			"failed to start watching vtxos",
-		)
-	}
-	log.Debugf("started watching %d vtxos", len(newVtxos))
+		if err := s.repoManager.Vtxos().SpendVtxos(ctx, spentVtxoKeys, redeemTxid); err != nil {
+			log.WithError(err).Warn("failed to spend vtxos")
+			return
+		}
+		log.Debugf("spent %d vtxos", len(spentVtxos))
 
-	if err := s.repoManager.Vtxos().SpendVtxos(ctx, spentVtxoKeys, redeemTxid); err != nil {
-		return "", "", fmt.Errorf("failed to spend vtxo: %s", err)
-	}
-	log.Infof("spent %d vtxos", len(spentVtxos))
+		if err := s.startWatchingVtxos(newVtxos); err != nil {
+			log.WithError(err).Warn("failed to start watching vtxos")
+		} else {
+			log.Debugf("started watching %d vtxos", len(newVtxos))
+		}
 
-	go func() {
 		s.transactionEventsCh <- RedeemTransactionEvent{
 			RedeemTxid:     redeemTxid,
 			SpentVtxos:     spentVtxos,
 			SpendableVtxos: newVtxos,
 			TxHex:          signedRedeemTx,
 		}
-	}()
+	}(ptx, signedRedeemTx, redeemTxid)
 
 	return signedRedeemTx, redeemTxid, nil
 }
@@ -1179,20 +1178,23 @@ func (s *covenantlessService) RegisterCosignerNonces(
 	if err != nil {
 		return fmt.Errorf("failed to decode nonces: %s", err)
 	}
-	session.lock.Lock()
-	defer session.lock.Unlock()
 
-	if _, ok := session.nonces[pubkey]; ok {
-		return nil // skip if we already have nonces for this pubkey
-	}
+	go func(session *musigSigningSession) {
+		session.lock.Lock()
+		defer session.lock.Unlock()
 
-	session.nonces[pubkey] = nonces
+		if _, ok := session.nonces[pubkey]; ok {
+			return // skip if we already have nonces for this pubkey
+		}
 
-	if len(session.nonces) == session.nbCosigners-1 { // exclude the server
-		go func() {
-			session.nonceDoneC <- struct{}{}
-		}()
-	}
+		session.nonces[pubkey] = nonces
+
+		if len(session.nonces) == session.nbCosigners-1 { // exclude the server
+			go func() {
+				session.nonceDoneC <- struct{}{}
+			}()
+		}
+	}(session)
 
 	return nil
 }
@@ -1215,20 +1217,22 @@ func (s *covenantlessService) RegisterCosignerSignatures(
 		return fmt.Errorf("failed to decode signatures: %s", err)
 	}
 
-	session.lock.Lock()
-	defer session.lock.Unlock()
+	go func(session *musigSigningSession) {
+		session.lock.Lock()
+		defer session.lock.Unlock()
 
-	if _, ok := session.signatures[pubkey]; ok {
-		return nil // skip if we already have signatures for this pubkey
-	}
+		if _, ok := session.signatures[pubkey]; ok {
+			return // skip if we already have signatures for this pubkey
+		}
 
-	session.signatures[pubkey] = signatures
+		session.signatures[pubkey] = signatures
 
-	if len(session.signatures) == session.nbCosigners-1 { // exclude the server
-		go func() {
-			session.sigDoneC <- struct{}{}
-		}()
-	}
+		if len(session.signatures) == session.nbCosigners-1 { // exclude the server
+			go func() {
+				session.sigDoneC <- struct{}{}
+			}()
+		}
+	}(session)
 
 	return nil
 }
@@ -1248,13 +1252,18 @@ func (s *covenantlessService) SetNostrRecipient(ctx context.Context, nostrRecipi
 		vtxoKeys = append(vtxoKeys, signedVtxo.Outpoint)
 	}
 
-	return s.repoManager.Entities().Add(
-		ctx,
-		domain.Entity{
-			NostrRecipient: nprofileRecipient,
-		},
-		vtxoKeys,
-	)
+	entity := domain.Entity{NostrRecipient: nprofileRecipient}
+	go func(entity domain.Entity, vtxoKeys []domain.VtxoKey) {
+		ctx := context.Background()
+
+		if err := s.repoManager.Entities().Add(ctx, entity, vtxoKeys); err != nil {
+			log.WithError(err).Warn("failed to add nostr identity, retrying...")
+			return
+		}
+		log.Debug("added new nostr identity")
+	}(entity, vtxoKeys)
+
+	return nil
 }
 
 func (s *covenantlessService) DeleteNostrRecipient(ctx context.Context, signedVtxoOutpoints []SignedVtxoOutpoint) error {
@@ -1267,7 +1276,28 @@ func (s *covenantlessService) DeleteNostrRecipient(ctx context.Context, signedVt
 		vtxoKeys = append(vtxoKeys, signedVtxo.Outpoint)
 	}
 
-	return s.repoManager.Entities().Delete(ctx, vtxoKeys)
+	fetchedKeys := make([]domain.VtxoKey, 0, len(vtxoKeys))
+	for _, vtxoKey := range vtxoKeys {
+		entities, err := s.repoManager.Entities().Get(ctx, vtxoKey)
+		if err != nil || len(entities) == 0 {
+			continue
+		}
+		fetchedKeys = append(fetchedKeys, vtxoKey)
+	}
+	if len(fetchedKeys) <= 0 {
+		return nil
+	}
+
+	go func(vtxoKeys []domain.VtxoKey) {
+		ctx := context.Background()
+		if err := s.repoManager.Entities().Delete(ctx, vtxoKeys); err != nil {
+			log.WithError(err).Warn("failed to delete nostr identity, retrying...")
+			return
+		}
+		log.Debug("deleted nostr identities")
+	}(fetchedKeys)
+
+	return nil
 }
 
 func (s *covenantlessService) start() {

server/internal/infrastructure/db/badger/entity_repo.go

@@ -2,8 +2,10 @@ package badgerdb
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"path/filepath"
+	"time"
 
 	"github.com/ark-network/ark/server/internal/core/domain"
 	"github.com/dgraph-io/badger/v4"
@@ -72,9 +74,16 @@ func (r *entityRepository) Add(ctx context.Context, entity domain.Entity, vtxoKe
 			entities.Entities = append(entities.Entities, entity)
 		}
 
-		err = r.store.Upsert(vtxoKey.String(), entities)
-		if err != nil {
-			return fmt.Errorf("failed to upsert entity: %w", err)
+		if err := r.store.Upsert(vtxoKey.String(), entities); err != nil {
+			if errors.Is(err, badger.ErrConflict) {
+				attempts := 1
+				for errors.Is(err, badger.ErrConflict) && attempts <= maxRetries {
+					time.Sleep(100 * time.Millisecond)
+					err = r.store.Upsert(vtxoKey.String(), entities)
+					attempts++
+				}
+			}
+			return err
 		}
 	}
 	return nil
@@ -98,12 +107,19 @@ func (r *entityRepository) Delete(ctx context.Context, vtxoKeys []domain.VtxoKey
 	}
 
 	for _, vtxoKey := range vtxoKeys {
-		err := r.store.Delete(vtxoKey.String(), &entities{})
-		if err == badgerhold.ErrNotFound {
-			return fmt.Errorf("entities not found for key: %s", vtxoKey)
-		}
-		if err != nil {
-			return fmt.Errorf("failed to delete entity: %w", err)
+		if err := r.store.Delete(vtxoKey.String(), &entities{}); err != nil {
+			if errors.Is(err, badgerhold.ErrNotFound) {
+				continue
+			}
+			if errors.Is(err, badger.ErrConflict) {
+				attempts := 1
+				for errors.Is(err, badger.ErrConflict) && attempts <= maxRetries {
+					time.Sleep(100 * time.Millisecond)
+					err = r.store.Delete(vtxoKey.String(), &entities{})
+					attempts++
+				}
+			}
+			return err
 		}
 	}
 	return nil