ledger: convert FC unmarshalled nil value to empty byte slice on DB w…

…rite KVs (#5225)

ledger/acctdeltas_test.go

@@ -850,6 +850,7 @@ func TestLookupKeysByPrefix(t *testing.T) {
 		{key: []byte("DingHo-StandardPack"), value: []byte("5bucks25cents")},
 		{key: []byte("BostonKitchen-CheeseSlice"), value: []byte("3bucks50cents")},
 		{key: []byte(`™£´´∂ƒ∂ƒßƒ©∑®ƒß∂†¬∆`), value: []byte("random Bluh")},
+		{key: []byte(`a-random-box-key`), value: []byte{}},
 	}
 
 	err = dbs.Transaction(func(ctx context.Context, tx trackerdb.TransactionScope) (err error) {
@@ -1067,6 +1068,143 @@ func BenchmarkLookupKeyByPrefix(b *testing.B) {
 	}
 }
 
+func TestKVStoreNilBlobConversion(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	t.Parallel()
+
+	// +-------------------------------------------------------------+
+	// | Section 1: Create a ledger with tracer DB of user_version 9 |
+	// +-------------------------------------------------------------+
+
+	const inMem = false
+
+	log := logging.TestingLog(t)
+	log.SetLevel(logging.Info)
+
+	dbs, dbName := storetesting.DbOpenTest(t, inMem)
+	storetesting.SetDbLogging(t, dbs)
+
+	err := dbs.Wdb.Atomic(func(ctx context.Context, tx *sql.Tx) error {
+		sqlitedriver.AccountsInitTest(t, tx, make(map[basics.Address]basics.AccountData), protocol.ConsensusCurrentVersion)
+		return nil
+	})
+	require.NoError(t, err)
+
+	defer func() {
+		dbs.Close()
+		require.NoError(t, os.Remove(dbName))
+	}()
+
+	targetVersion := int32(10)
+
+	err = dbs.Wdb.Atomic(func(ctx context.Context, tx *sql.Tx) (err0 error) {
+		_, err0 = tx.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version = %d", targetVersion-1))
+		return
+	})
+	require.NoError(t, err)
+
+	// +-----------------------------------------------------------------+
+	// | ^ Section 1 finishes above                                      |
+	// |                                                                 |
+	// | Section 2: jams a bunch of key value with value nil into the DB |
+	// +-----------------------------------------------------------------+
+
+	kvPairDBPrepareSet := []struct{ key []byte }{
+		{key: []byte{0xFF, 0x12, 0x34, 0x56, 0x78}},
+		{key: []byte{0xFF, 0xFF, 0x34, 0x56, 0x78}},
+		{key: []byte{0xFF, 0xFF, 0xFF, 0x56, 0x78}},
+		{key: []byte{0xFF, 0xFF, 0xFF, 0xFF, 0x78}},
+		{key: []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF}},
+		{key: []byte{0xFF, 0xFE, 0xFF}},
+		{key: []byte{0xFF, 0xFF, 0x00, 0xFF, 0xFF}},
+		{key: []byte{0xFF, 0xFF}},
+		{key: []byte{0xBA, 0xDD, 0xAD, 0xFF, 0xFF}},
+		{key: []byte{0xBA, 0xDD, 0xAE, 0x00}},
+		{key: []byte{0xBA, 0xDD, 0xAE}},
+		{key: []byte("TACOCAT")},
+		{key: []byte("TACOBELL")},
+		{key: []byte("DingHo-SmallPack")},
+		{key: []byte("DingHo-StandardPack")},
+		{key: []byte("BostonKitchen-CheeseSlice")},
+		{key: []byte(`™£´´∂ƒ∂ƒßƒ©∑®ƒß∂†¬∆`)},
+	}
+
+	err = dbs.Wdb.Atomic(func(ctx context.Context, tx *sql.Tx) (err0 error) {
+		writer, err0 := sqlitedriver.MakeAccountsSQLWriter(tx, false, false, true, false)
+		defer writer.Close()
+		if err0 != nil {
+			return
+		}
+		for i := 0; i < len(kvPairDBPrepareSet); i++ {
+			err0 = writer.UpsertKvPair(string(kvPairDBPrepareSet[i].key), nil)
+			if err0 != nil {
+				return
+			}
+		}
+		return
+	})
+	require.NoError(t, err)
+
+	// +---------------------------------------------------------------------------+
+	// | ^ Section 2 finishes above                                                |
+	// |                                                                           |
+	// | Section 3: Confirm that tracker DB has value being nil, not anything else |
+	// +---------------------------------------------------------------------------+
+
+	nilRowCounter := func() (nilRowCount int, err error) {
+		err = dbs.Wdb.Atomic(func(ctx context.Context, tx *sql.Tx) (err0 error) {
+			stmt, err0 := tx.PrepareContext(ctx, "SELECT key FROM kvstore WHERE value IS NULL;")
+			if err0 != nil {
+				return
+			}
+			rows, err0 := stmt.QueryContext(ctx)
+			if err0 != nil {
+				return
+			}
+			for rows.Next() {
+				var key sql.NullString
+				if err0 = rows.Scan(&key); err0 != nil {
+					return
+				}
+				if !key.Valid {
+					err0 = fmt.Errorf("scan from db get invalid key: %#v", key)
+					return
+				}
+				nilRowCount++
+			}
+			return
+		})
+		return
+	}
+
+	nilRowCount, err := nilRowCounter()
+	require.NoError(t, err)
+	require.Equal(t, len(kvPairDBPrepareSet), nilRowCount)
+
+	// +---------------------------------------------------------------------+
+	// | ^ Section 3 finishes above                                          |
+	// |                                                                     |
+	// | Section 4: Run migration to see replace nils with empty byte slices |
+	// +---------------------------------------------------------------------+
+
+	trackerDBWrapper := sqlitedriver.CreateTrackerSQLStore(dbs)
+	err = trackerDBWrapper.Transaction(func(ctx context.Context, tx trackerdb.TransactionScope) (err0 error) {
+		_, err0 = tx.RunMigrations(ctx, trackerdb.Params{}, log, targetVersion)
+		return
+	})
+	require.NoError(t, err)
+
+	// +------------------------------------------------------------------------------------------------+
+	// | ^ Section 4 finishes above                                                                     |
+	// |                                                                                                |
+	// | After that, we can confirm the DB migration found all nil strings and executed the conversions |
+	// +------------------------------------------------------------------------------------------------+
+
+	nilRowCount, err = nilRowCounter()
+	require.NoError(t, err)
+	require.Equal(t, 0, nilRowCount)
+}
+
 // upsert updates existing or inserts a new entry
 func (a *compactResourcesDeltas) upsert(delta resourceDelta) {
 	if idx, exist := a.cache[accountCreatable{address: delta.address, index: delta.oldResource.Aidx}]; exist {

ledger/catchupaccessor.go

@@ -128,6 +128,31 @@ func (w *stagingWriterImpl) writeKVs(ctx context.Context, kvrs []encoded.KVRecor
 		hashes := make([][]byte, len(kvrs))
 		for i := 0; i < len(kvrs); i++ {
 			keys[i] = kvrs[i].Key
+
+			// Since `encoded.KVRecordV6` is `omitempty` and `omitemptyarray`,
+			// when we have an instance of `encoded.KVRecordV6` with nil value,
+			// an empty box is unmarshalled to have `nil` value,
+			// while this might be mistaken to be a box deletion.
+			//
+			// We don't want to mistake this to be a deleted box:
+			// We are (and should be) during Fast Catchup (FC)
+			// writing to DB with empty byte string, rather than writing nil.
+			//
+			// This matters in sqlite3,
+			// for sqlite3 differs on writing nil byte slice to table from writing []byte{}:
+			// - writing nil byte slice is true that `value is NULL`
+			// - writing []byte{} is false on `value is NULL`.
+			//
+			// For the sake of consistency, we convert nil to []byte{}.
+			//
+			// Also, from a round by round catchup perspective,
+			// when we delete a box, in accountsNewRoundImpl method,
+			// the kv pair with value = nil will be deleted from kvstore table.
+			// Thus, it seems more consistent and appropriate to write as []byte{}.
+
+			if kvrs[i].Value == nil {
+				kvrs[i].Value = []byte{}
+			}
 			values[i] = kvrs[i].Value
 			hashes[i] = trackerdb.KvHashBuilderV6(string(keys[i]), values[i])
 		}

ledger/store/trackerdb/sqlitedriver/schema.go

@@ -350,6 +350,12 @@ func accountsCreateBoxTable(ctx context.Context, tx *sql.Tx) error {
 	return nil
 }
 
+// performKVStoreNullBlobConversion scans keys with null blob value, and convert the value to `[]byte{}`.
+func performKVStoreNullBlobConversion(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, "UPDATE kvstore SET value = '' WHERE value is NULL")
+	return err
+}
+
 func accountsCreateTxTailTable(ctx context.Context, tx *sql.Tx) (err error) {
 	for _, stmt := range createTxTailTable {
 		_, err = tx.ExecContext(ctx, stmt)

ledger/store/trackerdb/sqlitedriver/testing.go

@@ -103,6 +103,9 @@ func AccountsInitTest(tb testing.TB, tx *sql.Tx, initAccounts map[basics.Address
 	err = accountsCreateBoxTable(context.Background(), tx)
 	require.NoError(tb, err)
 
+	err = performKVStoreNullBlobConversion(context.Background(), tx)
+	require.NoError(tb, err)
+
 	return newDB
 }
 

ledger/store/trackerdb/sqlitedriver/trackerdbV2.go

@@ -485,14 +485,20 @@ func (tu *trackerDBSchemaInitializer) upgradeDatabaseSchema8(ctx context.Context
 	return tu.setVersion(ctx, tx, 9)
 }
 
-// upgradeDatabaseSchema9 upgrades the database schema from version 8 to version 9,
-// adding a new stateproofverification table.
+// upgradeDatabaseSchema9 upgrades the database schema from version 9 to version 10,
+// adding a new stateproofverification table,
+// scrubbing out all nil values from kvstore table and replace with empty byte slice.
 func (tu *trackerDBSchemaInitializer) upgradeDatabaseSchema9(ctx context.Context, tx *sql.Tx) (err error) {
 	err = createStateProofVerificationTable(ctx, tx)
 	if err != nil {
 		return err
 	}
 
+	err = performKVStoreNullBlobConversion(ctx, tx)
+	if err != nil {
+		return fmt.Errorf("upgradeDatabaseSchema9 unable to replace kvstore nil entries with empty byte slices : %v", err)
+	}
+
 	// update version
 	return tu.setVersion(ctx, tx, 10)
 }