fix(accounts): fix bloom filter empty

When rebooting the ledger, bloom filter will be empty and we will never fetch the metadata and the balances
formancehq · Aug 31, 2023 · d49a642 · d49a642
1 parent 78a58db
commit d49a642
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 51 deletions.
diff --git a/pkg/storage/sqlstorage/accounts.go b/pkg/storage/sqlstorage/accounts.go
@@ -271,8 +271,6 @@ func (s *Store) ensureAccountExists(ctx context.Context, account string) error {
 		return err
 	}
 
-	s.bloom.Add([]byte(account))
-
 	_, err = executor.ExecContext(ctx, sqlq, args...)
 	return s.error(err)
 }
@@ -326,8 +324,6 @@ func (s *Store) UpdateAccountMetadata(ctx context.Context, address string, metad
 		return errors.Wrap(err, "reading last log")
 	}
 
-	s.bloom.Add([]byte(address))
-
 	return s.appendLog(ctx, core.NewSetMetadataLog(lastLog, at, core.SetMetadata{
 		TargetType: core.MetaTargetTypeAccount,
 		TargetID:   address,

diff --git a/pkg/storage/sqlstorage/aggregations.go b/pkg/storage/sqlstorage/aggregations.go
@@ -23,65 +23,63 @@ func (s *Store) GetAccountWithVolumes(ctx context.Context, address string) (*cor
 	}
 	assetsVolumes := core.AssetsVolumes{}
 
-	if s.bloom.Test([]byte(address)) {
-		sb := sqlbuilder.NewSelectBuilder()
-		sb.Select("accounts.metadata", "volumes.asset", "volumes.input", "volumes.output")
-		sb.From(s.schema.Table("accounts"))
-		sb.JoinWithOption(sqlbuilder.LeftOuterJoin,
-			s.schema.Table("volumes"),
-			"accounts.address = volumes.account")
-		sb.Where(sb.E("accounts.address", address))
-
-		executor, err := s.executorProvider(ctx)
-		if err != nil {
-			return nil, err
-		}
+	sb := sqlbuilder.NewSelectBuilder()
+	sb.Select("accounts.metadata", "volumes.asset", "volumes.input", "volumes.output")
+	sb.From(s.schema.Table("accounts"))
+	sb.JoinWithOption(sqlbuilder.LeftOuterJoin,
+		s.schema.Table("volumes"),
+		"accounts.address = volumes.account")
+	sb.Where(sb.E("accounts.address", address))
 
-		q, args := sb.BuildWithFlavor(s.schema.Flavor())
-		rows, err := executor.QueryContext(ctx, q, args...)
-		if err != nil {
+	executor, err := s.executorProvider(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	q, args := sb.BuildWithFlavor(s.schema.Flavor())
+	rows, err := executor.QueryContext(ctx, q, args...)
+	if err != nil {
+		return nil, s.error(err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var asset, inputStr, outputStr sql.NullString
+		if err := rows.Scan(&acc.Metadata, &asset, &inputStr, &outputStr); err != nil {
 			return nil, s.error(err)
 		}
-		defer rows.Close()
 
-		for rows.Next() {
-			var asset, inputStr, outputStr sql.NullString
-			if err := rows.Scan(&acc.Metadata, &asset, &inputStr, &outputStr); err != nil {
-				return nil, s.error(err)
+		if asset.Valid {
+			assetsVolumes[asset.String] = core.Volumes{
+				Input:  core.NewMonetaryInt(0),
+				Output: core.NewMonetaryInt(0),
 			}
 
-			if asset.Valid {
+			if inputStr.Valid {
+				input, err := core.ParseMonetaryInt(inputStr.String)
+				if err != nil {
+					return nil, s.error(err)
+				}
 				assetsVolumes[asset.String] = core.Volumes{
-					Input:  core.NewMonetaryInt(0),
-					Output: core.NewMonetaryInt(0),
+					Input:  input,
+					Output: assetsVolumes[asset.String].Output,
 				}
+			}
 
-				if inputStr.Valid {
-					input, err := core.ParseMonetaryInt(inputStr.String)
-					if err != nil {
-						return nil, s.error(err)
-					}
-					assetsVolumes[asset.String] = core.Volumes{
-						Input:  input,
-						Output: assetsVolumes[asset.String].Output,
-					}
+			if outputStr.Valid {
+				output, err := core.ParseMonetaryInt(outputStr.String)
+				if err != nil {
+					return nil, s.error(err)
 				}
-
-				if outputStr.Valid {
-					output, err := core.ParseMonetaryInt(outputStr.String)
-					if err != nil {
-						return nil, s.error(err)
-					}
-					assetsVolumes[asset.String] = core.Volumes{
-						Input:  assetsVolumes[asset.String].Input,
-						Output: output,
-					}
+				assetsVolumes[asset.String] = core.Volumes{
+					Input:  assetsVolumes[asset.String].Input,
+					Output: output,
 				}
 			}
 		}
-		if err := rows.Err(); err != nil {
-			return nil, s.error(err)
-		}
+	}
+	if err := rows.Err(); err != nil {
+		return nil, s.error(err)
 	}
 
 	res := &core.AccountWithVolumes{

diff --git a/pkg/storage/sqlstorage/store_ledger.go b/pkg/storage/sqlstorage/store_ledger.go
@@ -27,7 +27,6 @@ type Store struct {
 	onDelete         func(ctx context.Context) error
 	lastLog          *core.Log
 	lastTx           *core.ExpandedTransaction
-	bloom            *bloom.BloomFilter
 	cache            *cache.Cache
 }