fix: rabbit/arkana comments

Author: chris-ricketts

internal/infrastructure/db/postgres/asset_repo.go

@@ -143,9 +143,13 @@ func (r *assetRepository) GetAssets(
 			}
 
 			if row.Metadata.Valid {
-				ast.Metadata, err = asset.NewMetadataListFromString(row.Metadata.String)
-				if err != nil {
-					log.WithError(err).Warnf("failed to parse metadata for asset %s", row.ID)
+				// Parsing metadata should never fail but if it does we just return an empty list
+				// of metadata and log the error
+				metadata, parseErr := asset.NewMetadataListFromString(row.Metadata.String)
+				if parseErr != nil {
+					log.WithError(parseErr).Warnf("failed to parse metadata for asset %s", row.ID)
+				} else {
+					ast.Metadata = metadata
 				}
 			}
 

internal/infrastructure/db/sqlite/asset_repo.go

@@ -158,13 +158,9 @@ func (r *assetRepository) GetAssets(
 			indexByID[row.ID] = idx
 		}
 
-		if !row.AssetAmount.Valid {
-			continue
-		}
-
-		amount, ok := new(big.Int).SetString(row.AssetAmount.String, 10)
+		amount, ok := new(big.Int).SetString(row.AssetAmount, 10)
 		if !ok {
-			continue
+			return nil, fmt.Errorf("invalid supply value: %s", row.AssetAmount)
 		}
 		assets[idx].Supply.Add(&assets[idx].Supply, amount)
 	}

internal/infrastructure/db/sqlite/sqlc/queries/query.sql.go

@@ -453,7 +453,7 @@ SELECT
   a.metadata_hash,
   a.metadata,
   a.control_asset_id,
-  v.asset_amount
+  COALESCE(v.asset_amount, '0') AS asset_amount
 FROM asset a
 LEFT JOIN vtxo_vw v
   ON v.asset_id = a.id

internal/infrastructure/db/sqlite/sqlc/query.sql

@@ -13,7 +13,9 @@ import (
 	"time"
 
 	"github.com/arkade-os/arkd/internal/infrastructure/db/sqlite/sqlc/queries"
-	_ "modernc.org/sqlite"
+	log "github.com/sirupsen/logrus"
+	sqlite "modernc.org/sqlite"
+	sqlite3 "modernc.org/sqlite/lib"
 )
 
 const (
@@ -74,20 +76,33 @@ func OpenDb(dbPath string) (SQLiteDB, error) {
 	}
 	writeDB.SetMaxOpenConns(1)
 
-	// Force connections to open now.
-	if err := writeDB.Ping(); err != nil {
+	// Use WAL so reads do not block writes
+	if _, err := writeDB.Exec(`PRAGMA journal_mode = WAL;`); err != nil {
 		_ = readDB.Close()
 		_ = writeDB.Close()
-		return nil, fmt.Errorf("failed to ping write db: %w", err)
+		return nil, fmt.Errorf("failed to enable WAL: %w", err)
 	}
 
-	// Use WAL so reads do not block writes
-	if _, err := writeDB.Exec(`PRAGMA journal_mode = WAL;`); err != nil {
+	// Use busy_timeout so reads/writes wait for WAL checkpointing instead of failing
+	if _, err := writeDB.Exec(`PRAGMA busy_timeout = 5000;`); err != nil {
+		_ = readDB.Close()
+		_ = writeDB.Close()
+		return nil, fmt.Errorf("failed to enable WAL: %w", err)
+	}
+
+	if _, err := readDB.Exec(`PRAGMA busy_timeout = 5000;`); err != nil {
 		_ = readDB.Close()
 		_ = writeDB.Close()
 		return nil, fmt.Errorf("failed to enable WAL: %w", err)
 	}
 
+	// Check there are no errors when opening a connection
+	if err := writeDB.Ping(); err != nil {
+		_ = readDB.Close()
+		_ = writeDB.Close()
+		return nil, fmt.Errorf("failed to ping write db: %w", err)
+	}
+
 	if err := readDB.Ping(); err != nil {
 		_ = readDB.Close()
 		_ = writeDB.Close()
@@ -164,7 +179,7 @@ func execTx(
 		}
 
 		if err := closeConn(conn, false); err != nil {
-			return err
+			log.WithError(err).Warn("failed to close connection after successful commit")
 		}
 		return nil
 	}
@@ -250,9 +265,8 @@ func closeConn(conn *sql.Conn, discard bool) error {
 	return nil
 }
 
-// isInterruptError reports whether err looks like a context cancellation or a
-// SQLite interrupt. SQLite interrupt detection is partly heuristic because the
-// driver may surface it as driver-specific error text.
+// isInterruptError reports whether err is a context cancellation or a SQLite
+// interrupt.
 func isInterruptError(ctx context.Context, err error) bool {
 	if err == nil {
 		return false
@@ -266,8 +280,13 @@ func isInterruptError(ctx context.Context, err error) bool {
 		return true
 	}
 
-	errMsg := strings.ToLower(err.Error())
-	return strings.Contains(errMsg, "interrupted") || strings.Contains(errMsg, "sqlite_interrupt")
+	var sqliteErr *sqlite.Error
+	if errors.As(err, &sqliteErr) {
+		code := sqliteErr.Code()
+		return code == sqlite3.SQLITE_INTERRUPT || code&0xff == sqlite3.SQLITE_INTERRUPT
+	}
+
+	return false
 }
 
 func isConflictError(err error) bool {

internal/infrastructure/db/sqlite/utils.go

@@ -5568,6 +5568,8 @@ func TestGetAssetQueryChurn(t *testing.T) {
 	var aliceRecvd, bobRecvd []types.Vtxo
 
 	for i := range supply {
+		completed := i + 1
+
 		sendWg := &sync.WaitGroup{}
 		sendWg.Add(2)
 		recvWg := &sync.WaitGroup{}
@@ -5607,14 +5609,14 @@ func TestGetAssetQueryChurn(t *testing.T) {
 		}()
 
 		sendWg.Wait()
-		require.NoErrorf(t, aliceSendErr, "send %d/%d failed", i, supply)
-		require.NoErrorf(t, bobSendErr, "send %d/%d failed", i, supply)
+		require.NoErrorf(t, aliceSendErr, "send %d/%d failed", completed, supply)
+		require.NoErrorf(t, bobSendErr, "send %d/%d failed", completed, supply)
 
 		recvWg.Wait()
 		require.NoError(t, aliceRecvErr, "receiving vtxos for send %s %d/%d failed",
-			aliceSendRes.Txid, i, supply)
+			aliceSendRes.Txid, completed, supply)
 		require.NoError(t, bobRecvErr, "receiving vtxos for send %s %d/%d failed",
-			bobSendRes.Txid, i, supply)
+			bobSendRes.Txid, completed, supply)
 
 		outpoints := make([]types.Outpoint, 0)
 		spentVtxos := make([]types.Outpoint, 0)
@@ -5666,7 +5668,7 @@ func TestGetAssetQueryChurn(t *testing.T) {
 		}
 
 		// start a batch after every batchInterval sends
-		if i != 0 && i%batchInterval == 0 {
+		if completed%batchInterval == 0 {
 			settleWg := &sync.WaitGroup{}
 			settleWg.Add(4)
 
@@ -5735,7 +5737,7 @@ func TestGetAssetQueryChurn(t *testing.T) {
 			require.NoError(t, bobGetCtxErr)
 			require.Len(t, bobCtx.Batches, 1, "failed to update completed round in database")
 			t.Logf("completed %d/%d offchain sends and batch %d/%d",
-				i, supply, i/batchInterval, supply/batchInterval)
+				completed, supply, completed/batchInterval, supply/batchInterval)
 		}
 	}
 

internal/test/e2e/e2e_test.go

@@ -811,14 +811,18 @@ func isRetryableChurnError(err error) bool {
 	return false
 }
 
-func waitForVTXOs(ch <-chan indexer.ScriptEvent, atLeastN int, timeout time.Duration) ([]types.Vtxo, error) {
+func waitForVTXOs(
+	ch <-chan indexer.ScriptEvent,
+	atLeastN int,
+	timeout time.Duration,
+) ([]types.Vtxo, error) {
 	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(timeout))
 	defer cancel()
 	vtxos := make([]types.Vtxo, 0)
 	for {
 		select {
 		case <-ctx.Done():
-			return nil, fmt.Errorf("timed out - %d/%d recieved", len(vtxos), atLeastN)
+			return nil, fmt.Errorf("timed out - %d/%d received", len(vtxos), atLeastN)
 		case evt, ok := <-ch:
 			if !ok {
 				return nil, fmt.Errorf("vtxo event channel closed")