Merge branch 'master' into release-horizon-v2.30.0

Makefile

@@ -19,7 +19,7 @@ XDRS = $(DOWNLOADABLE_XDRS) xdr/Stellar-lighthorizon.x \
 
 
 XDRGEN_COMMIT=e2cac557162d99b12ae73b846cf3d5bfe16636de
-XDR_COMMIT=59062438237d5f77fd6feb060b76288e88b7e222
+XDR_COMMIT=2ba4049554bb0564950e6d9213e01a60fc190f54
 
 .PHONY: xdr xdr-clean xdr-update
 

services/horizon/internal/app.go

@@ -512,7 +512,11 @@ func (a *App) init() error {
 	initSubmissionSystem(a)
 
 	// reaper
-	a.reaper = reap.New(a.config.HistoryRetentionCount, a.HorizonSession(), a.ledgerState)
+	a.reaper = reap.New(
+		a.config.HistoryRetentionCount,
+		a.config.HistoryRetentionReapCount,
+		a.HorizonSession(),
+		a.ledgerState)
 
 	// go metrics
 	initGoMetrics(a)

services/horizon/internal/config.go

@@ -74,6 +74,12 @@ type Config struct {
 	// determining a "retention duration", each ledger roughly corresponds to 10
 	// seconds of real time.
 	HistoryRetentionCount uint
+	// HistoryRetentionReapCount is the number of ledgers worth of history data
+	// to remove per second from the Horizon database. It is intended to allow
+	// control over the amount of CPU and database load caused by reaping,
+	// especially if enabling reaping for the first time or in times of
+	// increased ledger load.
+	HistoryRetentionReapCount uint
 	// StaleThreshold represents the number of ledgers a history database may be
 	// out-of-date by before horizon begins to respond with an error to history
 	// requests.

services/horizon/internal/db2/history/reap_test.go

@@ -17,7 +17,7 @@ func TestReapLookupTables(t *testing.T) {
 
 	db := tt.HorizonSession()
 
-	sys := reap.New(0, db, ledgerState)
+	sys := reap.New(0, 0, db, ledgerState)
 
 	var (
 		prevLedgers, curLedgers                     int
@@ -43,6 +43,7 @@ func TestReapLookupTables(t *testing.T) {
 
 	ledgerState.SetStatus(tt.LoadLedgerStatus())
 	sys.RetentionCount = 1
+	sys.RetentionBatch = 50
 	err := sys.DeleteUnretainedHistory(tt.Ctx)
 	tt.Require.NoError(err)
 

services/horizon/internal/flags.go

@@ -650,9 +650,23 @@ func Flags() (*Config, support.ConfigOptions) {
 			ConfigKey:      &config.HistoryRetentionCount,
 			OptType:        types.Uint,
 			FlagDefault:    uint(0),
-			Usage:          "the minimum number of ledgers to maintain within horizon's history tables.  0 signifies an unlimited number of ledgers will be retained",
+			Usage:          "the minimum number of ledgers to maintain within Horizon's history tables (0 = retain an unlimited number of ledgers)",
 			UsedInCommands: IngestionCommands,
 		},
+		&support.ConfigOption{
+			Name:           "history-retention-reap-count",
+			ConfigKey:      &config.HistoryRetentionReapCount,
+			OptType:        types.Uint,
+			FlagDefault:    uint(50_000),
+			Usage:          "the batch size (in ledgers) to remove per reap from the Horizon database",
+			UsedInCommands: IngestionCommands,
+			CustomSetValue: func(opt *support.ConfigOption) error {
+				if val := viper.GetUint(opt.Name); val <= 0 || val > 500_000 {
+					return fmt.Errorf("flag --history-retention-reap-count must be in range [1, 500,000]")
+				}
+				return nil
+			},
+		},
 		&support.ConfigOption{
 			Name:           "history-stale-threshold",
 			ConfigKey:      &config.StaleThreshold,

services/horizon/internal/httpt_test.go

@@ -104,6 +104,7 @@ func (ht *HTTPT) Post(
 // setting the retention count to the provided number.
 func (ht *HTTPT) ReapHistory(retention uint) {
 	ht.App.reaper.RetentionCount = retention
+	ht.App.reaper.RetentionBatch = 50_000
 	ht.App.reaper.HistoryQ = &history.Q{ht.HorizonSession()}
 	err := ht.App.DeleteUnretainedHistory(context.Background())
 	ht.Require.NoError(err)

services/horizon/internal/reap/main.go

@@ -16,19 +16,22 @@ import (
 type System struct {
 	HistoryQ       *history.Q
 	RetentionCount uint
-	ledgerState    *ledger.State
-	ctx            context.Context
-	cancel         context.CancelFunc
+	RetentionBatch uint
+
+	ledgerState *ledger.State
+	ctx         context.Context
+	cancel      context.CancelFunc
 }
 
 // New initializes the reaper, causing it to begin polling the stellar-core
 // database for now ledgers and ingesting data into the horizon database.
-func New(retention uint, dbSession db.SessionInterface, ledgerState *ledger.State) *System {
+func New(retention, retentionBatchSize uint, dbSession db.SessionInterface, ledgerState *ledger.State) *System {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	r := &System{
 		HistoryQ:       &history.Q{dbSession.Clone()},
 		RetentionCount: retention,
+		RetentionBatch: retentionBatchSize,
 		ledgerState:    ledgerState,
 		ctx:            ctx,
 		cancel:         cancel,

services/horizon/internal/reap/system.go

@@ -2,6 +2,7 @@ package reap
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	herrors "github.com/stellar/go/services/horizon/internal/errors"
@@ -70,25 +71,31 @@ func (r *System) runOnce(ctx context.Context) {
 	}
 }
 
-// Work backwards in 100k ledger blocks to prevent using all the CPU.
+// Work backwards in 50k (by default, otherwise configurable via the CLI) ledger
+// blocks to prevent using all the CPU.
 //
-// This runs every hour, so we need to make sure it doesn't
-// run for longer than an hour.
+// This runs every hour, so we need to make sure it doesn't run for longer than
+// an hour.
 //
-// Current ledger at 2021-08-12 is 36,827,497, so 100k means 368 batches. At 1
-// batch/second, that seems like a reasonable balance between running well
-// under an hour, and slowing it down enough to leave some CPU for other
-// processes.
-var batchSize = int32(100_000)
+// Current ledger at 2024-04-04s is 51,092,283, so 50k means 1021 batches. At 1
+// batch/second, that seems like a reasonable balance between running under an
+// hour, and slowing it down enough to leave some CPU for other processes.
 var sleep = 1 * time.Second
 
 func (r *System) clearBefore(ctx context.Context, startSeq, endSeq int32) error {
+	batchSize := int32(r.RetentionBatch)
+	if batchSize <= 0 {
+		return fmt.Errorf("invalid batch size for reaping (%d)", batchSize)
+	}
+
 	for batchEndSeq := endSeq - 1; batchEndSeq >= startSeq; batchEndSeq -= batchSize {
 		batchStartSeq := batchEndSeq - batchSize
 		if batchStartSeq < startSeq {
 			batchStartSeq = startSeq
 		}
-		log.WithField("start_ledger", batchStartSeq).WithField("end_ledger", batchEndSeq).Info("reaper: clearing")
+		log.WithField("start_ledger", batchStartSeq).
+			WithField("end_ledger", batchEndSeq).
+			Info("reaper: clearing")
 
 		batchStart, batchEnd, err := toid.LedgerRangeInclusive(batchStartSeq, batchEndSeq)
 		if err != nil {

services/horizon/internal/reap/system_test.go

@@ -15,7 +15,7 @@ func TestDeleteUnretainedHistory(t *testing.T) {
 
 	db := tt.HorizonSession()
 
-	sys := New(0, db, ledgerState)
+	sys := New(0, 50, db, ledgerState)
 
 	// Disable sleeps for this.
 	sleep = 0

xdr/Stellar-contract-spec.x

@@ -42,6 +42,7 @@ enum SCSpecType
     SC_SPEC_TYPE_MAP = 1004,
     SC_SPEC_TYPE_TUPLE = 1005,
     SC_SPEC_TYPE_BYTES_N = 1006,
+    SC_SPEC_TYPE_HASH = 1007,
 
     // User defined types.
     SC_SPEC_TYPE_UDT = 2000
@@ -79,6 +80,11 @@ struct SCSpecTypeBytesN
     uint32 n;
 };
 
+struct SCSpectTypeHash
+{
+    uint32 n;
+};
+
 struct SCSpecTypeUDT
 {
     string name<60>;
@@ -117,6 +123,8 @@ case SC_SPEC_TYPE_TUPLE:
     SCSpecTypeTuple tuple;
 case SC_SPEC_TYPE_BYTES_N:
     SCSpecTypeBytesN bytesN;
+case SC_SPEC_TYPE_HASH:
+    SCSpectTypeHash hash;
 case SC_SPEC_TYPE_UDT:
     SCSpecTypeUDT udt;
 };

xdr/ledger_close_meta.go

@@ -19,6 +19,10 @@ func (l LedgerCloseMeta) LedgerSequence() uint32 {
 	return uint32(l.LedgerHeaderHistoryEntry().Header.LedgerSeq)
 }
 
+func (l LedgerCloseMeta) LedgerCloseTime() int64 {
+	return int64(l.LedgerHeaderHistoryEntry().Header.ScpValue.CloseTime)
+}
+
 func (l LedgerCloseMeta) LedgerHash() Hash {
 	return l.LedgerHeaderHistoryEntry().Hash
 }

xdr/xdr_commit_generated.txt

@@ -1 +1 @@
-59062438237d5f77fd6feb060b76288e88b7e222
+2ba4049554bb0564950e6d9213e01a60fc190f54