services/horizon/internal/ingest: Remove unnecessary use of ChangeCompactor to reduce memory bloat during ingestion

ingest/change.go

@@ -229,74 +229,3 @@ func (c *Change) AccountChangedExceptSigners() (bool, error) {
 
 	return !bytes.Equal(preBinary, postBinary), nil
 }
-
-// AccountSignersChanged returns true if account signers have changed.
-// Notice: this will return true on master key changes too!
-func (c *Change) AccountSignersChanged() bool {
-	if c.Type != xdr.LedgerEntryTypeAccount {
-		panic("This should not be called on changes other than Account changes")
-	}
-
-	// New account so new master key (which is also a signer)
-	if c.Pre == nil {
-		return true
-	}
-
-	// Account merged. Account being merge can still have signers.
-	// c.Pre != nil at this point.
-	if c.Post == nil {
-		return true
-	}
-
-	// c.Pre != nil && c.Post != nil at this point.
-	preAccountEntry := c.Pre.Data.MustAccount()
-	postAccountEntry := c.Post.Data.MustAccount()
-
-	preSigners := preAccountEntry.SignerSummary()
-	postSigners := postAccountEntry.SignerSummary()
-
-	if len(preSigners) != len(postSigners) {
-		return true
-	}
-
-	for postSigner, postWeight := range postSigners {
-		preWeight, exist := preSigners[postSigner]
-		if !exist {
-			return true
-		}
-
-		if preWeight != postWeight {
-			return true
-		}
-	}
-
-	preSignerSponsors := preAccountEntry.SignerSponsoringIDs()
-	postSignerSponsors := postAccountEntry.SignerSponsoringIDs()
-
-	if len(preSignerSponsors) != len(postSignerSponsors) {
-		return true
-	}
-
-	for i := 0; i < len(preSignerSponsors); i++ {
-		preSponsor := preSignerSponsors[i]
-		postSponsor := postSignerSponsors[i]
-
-		if preSponsor == nil && postSponsor != nil {
-			return true
-		} else if preSponsor != nil && postSponsor == nil {
-			return true
-		} else if preSponsor != nil && postSponsor != nil {
-			preSponsorAccountID := xdr.AccountId(*preSponsor)
-			preSponsorAddress := preSponsorAccountID.Address()
-
-			postSponsorAccountID := xdr.AccountId(*postSponsor)
-			postSponsorAddress := postSponsorAccountID.Address()
-
-			if preSponsorAddress != postSponsorAddress {
-				return true
-			}
-		}
-	}
-
-	return false
-}

ingest/change_compactor.go

@@ -14,6 +14,13 @@ import (
 // previously removed returns an error. In such case verify.StateError is
 // returned.
 //
+// The ChangeCompactor should not be used when ingesting from history archives
+// because the history archive snapshots only contain CREATED changes.
+// The ChangeCompactor is suited for compacting ledger entry changes derived
+// from LedgerCloseMeta payloads because they typically contain a mix of
+// CREATED, UPDATED, and REMOVED ledger entry changes and therefore may benefit
+// from compaction.
+//
 // It applies changes to the cache using the following algorithm:
 //
 //  1. If the change is CREATED it checks if any change connected to given entry

ingest/ledger_change_reader.go

@@ -76,6 +76,57 @@ func NewLedgerChangeReaderFromLedgerCloseMeta(networkPassphrase string, ledger x
 	}, nil
 }
 
+type compactingChangeReader struct {
+	input     ChangeReader
+	changes   []Change
+	compacted bool
+}
+
+func (c *compactingChangeReader) compact() error {
+	compactor := NewChangeCompactor()
+	for {
+		change, err := c.input.Read()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return err
+		}
+		if err = compactor.AddChange(change); err != nil {
+			return err
+		}
+	}
+	c.changes = compactor.GetChanges()
+	c.compacted = true
+	return nil
+}
+
+func (c *compactingChangeReader) Read() (Change, error) {
+	if !c.compacted {
+		if err := c.compact(); err != nil {
+			return Change{}, err
+		}
+	}
+	if len(c.changes) == 0 {
+		return Change{}, io.EOF
+	}
+	change := c.changes[0]
+	c.changes = c.changes[1:]
+	return change, nil
+}
+
+func (c *compactingChangeReader) Close() error {
+	return c.input.Close()
+}
+
+// NewCompactingChangeReader wraps a given ChangeReader and returns a ChangeReader
+// which compacts all the the Changes extracted from the input.
+func NewCompactingChangeReader(input ChangeReader) ChangeReader {
+	return &compactingChangeReader{
+		input: input,
+	}
+}
+
 // Read returns the next change in the stream.
 // If there are no changes remaining io.EOF is returned as an error.
 func (r *LedgerChangeReader) Read() (Change, error) {