services/horizon: Insert signers's sponsorship information during state ingestion (#3096)

Fixes `SignersProcessor` to insert sponsorship information during state
ingestion from history archives.

`SignersProcessor` operates in two modes: `useLedgerEntryCache` set to
`true` or `false`. When ingesting history archives we use
`useLedgerEntryCache=false` to save memory - ledger entry cache is not
needed because there are only inserts. This code path was not updated to
include information about the sponsor of a signer.

Author: bartekn

Diff for: services/horizon/internal/expingest/processors/signer_processor_test.go

@@ -4,6 +4,7 @@ package processors
 import (
 	"testing"
 
+	"github.com/guregu/null"
 	ingesterrors "github.com/stellar/go/exp/ingest/errors"
 	"github.com/stellar/go/exp/ingest/io"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
@@ -98,6 +99,53 @@ func (s *AccountsSignerProcessorTestSuiteState) TestCreatesSigners() {
 
 }
 
+func (s *AccountsSignerProcessorTestSuiteState) TestCreatesSignerWithSponsor() {
+	s.mockBatchInsertBuilder.
+		On("Add", history.AccountSigner{
+			Account: "GCCCU34WDY2RATQTOOQKY6SZWU6J5DONY42SWGW2CIXGW4LICAGNRZKX",
+			Signer:  "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+			Weight:  int32(10),
+			Sponsor: null.StringFrom("GDWZ6MKJP5ESVIB7O5RW4UFFGSCDILPEKDXWGG4HXXSHEZZPTKLR6UVG"),
+		}).Return(nil).Once()
+
+	sponsorshipDescriptor := xdr.MustAddress("GDWZ6MKJP5ESVIB7O5RW4UFFGSCDILPEKDXWGG4HXXSHEZZPTKLR6UVG")
+
+	err := s.processor.ProcessChange(io.Change{
+		Type: xdr.LedgerEntryTypeAccount,
+		Pre:  nil,
+		Post: &xdr.LedgerEntry{
+			Data: xdr.LedgerEntryData{
+				Type: xdr.LedgerEntryTypeAccount,
+				Account: &xdr.AccountEntry{
+					AccountId: xdr.MustAddress("GCCCU34WDY2RATQTOOQKY6SZWU6J5DONY42SWGW2CIXGW4LICAGNRZKX"),
+					Signers: []xdr.Signer{
+						{
+							Key:    xdr.MustSigner("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
+							Weight: 10,
+						},
+					},
+					Ext: xdr.AccountEntryExt{
+						V: 1,
+						V1: &xdr.AccountEntryExtensionV1{
+							Ext: xdr.AccountEntryExtensionV1Ext{
+								V: 2,
+								V2: &xdr.AccountEntryExtensionV2{
+									NumSponsored:  1,
+									NumSponsoring: 0,
+									SignerSponsoringIDs: []xdr.SponsorshipDescriptor{
+										&sponsorshipDescriptor,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	})
+	s.Assert().NoError(err)
+}
+
 func TestAccountsSignerProcessorTestSuiteLedger(t *testing.T) {
 	suite.Run(t, new(AccountsSignerProcessorTestSuiteLedger))
 }

Diff for: services/horizon/internal/expingest/processors/signers_processor.go

@@ -1,6 +1,7 @@
 package processors
 
 import (
+	"github.com/guregu/null"
 	ingesterrors "github.com/stellar/go/exp/ingest/errors"
 	"github.com/stellar/go/exp/ingest/io"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
@@ -61,11 +62,18 @@ func (p *SignersProcessor) ProcessChange(change io.Change) error {
 	accountEntry := change.Post.Data.MustAccount()
 	account := accountEntry.AccountId.Address()
 
+	sponsors := accountEntry.SponsorPerSigner()
 	for signer, weight := range accountEntry.SignerSummary() {
+		var sponsor null.String
+		if sponsorDesc, isSponsored := sponsors[signer]; isSponsored {
+			sponsor = null.StringFrom(sponsorDesc.Address())
+		}
+
 		err := p.batch.Add(history.AccountSigner{
 			Account: account,
 			Signer:  signer,
 			Weight:  weight,
+			Sponsor: sponsor,
 		})
 		if err != nil {
 			return errors.Wrap(err, "Error adding row to accountSignerBatch")