Merge upstream master into horizon-v2.28.2 to facilitate the back-merge

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2
 	github.com/aws/aws-sdk-go v1.45.26
 	github.com/creachadair/jrpc2 v1.1.0
+	github.com/djherbis/fscache v0.10.1
 	github.com/elazarl/go-bindata-assetfs v1.0.1
 	github.com/getsentry/raven-go v0.2.0
 	github.com/go-chi/chi v4.1.2+incompatible
@@ -91,6 +92,8 @@ require (
 	golang.org/x/tools v0.14.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20231211222908-989df2bf70f3 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240102182953-50ed04b92917 // indirect
+	gopkg.in/djherbis/atime.v1 v1.0.0 // indirect
+	gopkg.in/djherbis/stream.v1 v1.3.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
 

go.sum

@@ -104,6 +104,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/djherbis/fscache v0.10.1 h1:hDv+RGyvD+UDKyRYuLoVNbuRTnf2SrA2K3VyR1br9lk=
+github.com/djherbis/fscache v0.10.1/go.mod h1:yyPYtkNnnPXsW+81lAcQS6yab3G2CRfnPLotBvtbf0c=
 github.com/eapache/go-resiliency v1.2.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
@@ -814,6 +816,10 @@ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/djherbis/atime.v1 v1.0.0 h1:eMRqB/JrLKocla2PBPKgQYg/p5UG4L6AUAs92aP7F60=
+gopkg.in/djherbis/atime.v1 v1.0.0/go.mod h1:hQIUStKmJfvf7xdh/wtK84qe+DsTV5LnA9lzxxtPpJ8=
+gopkg.in/djherbis/stream.v1 v1.3.1 h1:uGfmsOY1qqMjQQphhRBSGLyA9qumJ56exkRu9ASTjCw=
+gopkg.in/djherbis/stream.v1 v1.3.1/go.mod h1:aEV8CBVRmSpLamVJfM903Npic1IKmb2qS30VAZ+sssg=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/gavv/httpexpect.v1 v1.0.0-20170111145843-40724cf1e4a0 h1:r5ptJ1tBxVAeqw4CrYWhXIMr0SybY3CDHuIbCg5CFVw=

historyarchive/archive.go

@@ -11,12 +11,15 @@
 	"fmt"
 	"io"
 	"net/url"
+	"os"
 	"path"
 	"regexp"
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
+	fscache "github.com/djherbis/fscache"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/stellar/go/support/errors"
@@ -38,16 +41,17 @@
 }
 
 type ArchiveOptions struct {
+	storage.ConnectOptions
+
 	// NetworkPassphrase defines the expected network of history archive. It is
 	// checked when getting HAS. If network passphrase does not match, error is
 	// returned.
 	NetworkPassphrase string
 	// CheckpointFrequency is the number of ledgers between checkpoints
 	// if unset, DefaultCheckpointFrequency will be used
 	CheckpointFrequency uint32
-	storage.ConnectOptions
-	// CacheConfig controls how/if bucket files are cached on the disk.
-	CacheConfig CacheOptions
+	// CachePath controls where/if bucket files are cached on the disk.
+	CachePath string
 }
 
 type Ledger struct {
@@ -104,8 +108,16 @@
 	checkpointManager CheckpointManager
 
 	backend storage.Storage
-	cache   *ArchiveBucketCache
 	stats   archiveStats
+
+	cache *archiveBucketCache
+}
+
+type archiveBucketCache struct {
+	fscache.Cache
+
+	path  string
+	sizes sync.Map
 }
 
 func (arch *Archive) GetStats() []ArchiveStats {
@@ -119,8 +131,9 @@
 func (a *Archive) GetPathHAS(path string) (HistoryArchiveState, error) {
 	var has HistoryArchiveState
 	rdr, err := a.backend.GetFile(path)
-	// this is a query on the HA server state, not a data/bucket file download
-	a.stats.incrementRequests()
+	a.stats.incrementDownloads()
+	// 	// this is a query on the HA server state, not a data/bucket file download
+	// 	a.stats.incrementRequests()
 	if err != nil {
 		return has, err
 	}
@@ -383,23 +396,79 @@
 }
 
 func (a *Archive) cachedGet(pth string) (io.ReadCloser, error) {
-	if a.cache != nil {
-		rdr, foundInCache, err := a.cache.GetFile(pth, a.backend)
-		if !foundInCache {
-			a.stats.incrementDownloads()
-		} else {
-			a.stats.incrementCacheHits()
-		}
-		if err == nil {
-			return rdr, nil
+	if a.cache == nil {
+		a.stats.incrementDownloads()
+		return a.backend.GetFile(pth)
+	}
+
+	L := log.WithField("path", pth).WithField("cache", a.cache.path)
+
+	rdr, wrtr, err := a.cache.Get(pth)
+	if err != nil {
+		L.WithError(err).
+			WithField("remove", a.cache.Remove(pth)).
+			Warn("On-disk cache retrieval failed")
+		a.stats.incrementDownloads()
+		return a.backend.GetFile(pth)
+	}
+
+	// If a NEW key is being retrieved, it returns a writer to which
+	// you're expected to write your upstream as well as a reader that
+	// will read directly from it.
+	if wrtr != nil {
+		log.WithField("path", pth).Info("Caching file...")
+		a.stats.incrementDownloads()
+		upstreamReader, err := a.backend.GetFile(pth)
+		if err != nil {
+			writeErr := wrtr.Close()
+			readErr := rdr.Close()
+			removeErr := a.cache.Remove(pth)
+			// Execution order isn't guaranteed w/in a function call expression
+			// so we close them with explicit order first.
+			L.WithError(err).WithFields(log.Fields{
+				"write-close": writeErr,
+				"read-close":  readErr,
+				"cache-rm":    removeErr,
+			}).Warn("Download failed, purging from cache")
+			return nil, err
 		}
 
-		// If there's an error, retry with the uncached backend.
-		a.cache.Evict(pth)
+		// Start a goroutine to slurp up the upstream and feed
+		// it directly to the cache.
+		go func() {
+			written, err := io.Copy(wrtr, upstreamReader)
+			writeErr := wrtr.Close()
+			readErr := upstreamReader.Close()
+			fields := log.Fields{
+				"wr-close": writeErr,
+				"rd-close": readErr,
+			}
+
+			if err != nil {
+				L.WithFields(fields).WithError(err).
+					Warn("Failed to download and cache file")
+
+				// Removal must happen *after* handles close.
+				if removalErr := a.cache.Remove(pth); removalErr != nil {
+					L.WithError(removalErr).Warn("Removing cached file failed")
+				}
+			} else {
+				L.WithFields(fields).Infof("Cached %dKiB file", written/1024)
+
+				// Track how much bandwidth we've saved from caching by saving
+				// the size of the file we just downloaded.
+				a.cache.sizes.Store(pth, written)
+			}
+		}()
+	} else {
+		// Best-effort check to track bandwidth metrics
+		if written, found := a.cache.sizes.Load(pth); found {
+			a.stats.incrementCacheBandwidth(written.(int64))
+		}
+		a.stats.incrementCacheHits()
 	}
 
-	a.stats.incrementDownloads()
-	return a.backend.GetFile(pth)
+	return rdr, nil
 }
 
 func (a *Archive) cachedExists(pth string) (bool, error) {
@@ -439,13 +508,30 @@
 		return &arch, err
 	}
 
-	if opts.CacheConfig.Cache {
-		cache, innerErr := MakeArchiveBucketCache(opts.CacheConfig)
-		if innerErr != nil {
-			return &arch, innerErr
+	if opts.CachePath != "" {
+		// Set up a <= ~10GiB LRU cache for history archives files
+		haunter := fscache.NewLRUHaunterStrategy(
+			fscache.NewLRUHaunter(0, 10<<30, time.Minute /* frequency check */),
+		)
+
+		// Wipe any existing cache on startup
+		os.RemoveAll(opts.CachePath)
+		fs, err := fscache.NewFs(opts.CachePath, 0755 /* drwxr-xr-x */)
+
+		if err != nil {
+			return &arch, errors.Wrapf(err,
+				"creating cache at '%s' with mode 0755 failed",
+				opts.CachePath)
+		}
+
+		cache, err := fscache.NewCacheWithHaunter(fs, haunter)
+		if err != nil {
+			return &arch, errors.Wrapf(err,
+				"creating cache at '%s' failed",
+				opts.CachePath)
 		}
 
-		arch.cache = cache
+		arch.cache = &archiveBucketCache{cache, opts.CachePath, sync.Map{}}
 	}
 
 	arch.stats = archiveStats{backendName: u}
@@ -467,6 +553,8 @@
 
 	if parsed.Scheme == "mock" {
 		backend = makeMockBackend()
+	} else if parsed.Scheme == "fmock" {
+		backend = makeFailingMockBackend()
 	} else {
 		backend, err = storage.ConnectBackend(u, opts)
 	}

historyarchive/archive_pool.go

@@ -31,11 +31,7 @@ func NewArchivePool(archiveURLs []string, opts ArchiveOptions) (ArchivePool, err
 	// Try connecting to all of the listed archives, but only store valid ones.
 	var validArchives ArchivePool
 	for _, url := range archiveURLs {
-		archive, err := Connect(
-			url,
-			opts,
-		)
-
+		archive, err := Connect(url, opts)
 		if err != nil {
 			lastErr = errors.Wrapf(err, "Error connecting to history archive (%s)", url)
 			continue