Improve feeders

cmd/feedwitness/main.go

@@ -26,19 +26,19 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"net"
 	"net/http"
 	"net/url"
-	"regexp"
 	"strings"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	w_http "github.com/transparency-dev/witness/client/http"
+	"github.com/transparency-dev/witness/internal/feeder"
 	"github.com/transparency-dev/witness/internal/witness"
 	"github.com/transparency-dev/witness/monitoring"
 	"github.com/transparency-dev/witness/monitoring/prometheus"
 	"github.com/transparency-dev/witness/omniwitness"
-	"golang.org/x/sync/errgroup"
-	"golang.org/x/time/rate"
 	"k8s.io/klog/v2"
 )
 
@@ -50,8 +50,7 @@ var (
 	witnessURL    multiStringFlag
 	httpsInsecure = flag.Bool("https_insecure", false, "Set to true to disable TLS verification of the witness service")
 	feed          = flag.String("feed", ".*", "RegEx matching log origins to feed checkpoints from")
-	loopInterval  = flag.Duration("loop_interval", 0, "If set to > 0, runs in looping mode sleeping this duration between feed attempts")
-	rateLimit     = flag.Float64("max_qps", 2, "Defines maximum number of requests/s to send")
+	rateLimit     = flag.Float64("max_qps", 2, "Defines maximum number of requests/s to send per witness")
 	metricsAddr   = flag.String("metrics_listen", ":8081", "Address to listen on for metrics")
 )
 
@@ -61,21 +60,12 @@ func main() {
 	defer klog.Flush()
 
 	ctx := context.Background()
-	rl := rate.NewLimiter(rate.Limit(*rateLimit), 1)
 
-	httpClient := http.DefaultClient
-	if *httpsInsecure {
-		httpClient = &http.Client{Transport: &http.Transport{
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-		},
-		}
-	}
 	cfg, err := omniwitness.NewStaticLogConfig(omniwitness.DefaultConfigLogs)
 	if err != nil {
 		klog.Exitf("failed to instantiate default witness config: %v", err)
 	}
 
-	r := regexp.MustCompile(*feed)
 	if len(witnessURL) == 0 {
 		klog.Exitf("At least one --witness_url must be specifed")
 	}
@@ -95,31 +85,29 @@ func main() {
 		klog.Infof("Prometheus configured to listen on %q", *metricsAddr)
 	}
 
-	eg := errgroup.Group{}
+	httpClient := httpClientFromFlags()
+
+	witnesses := []feeder.UpdateFn{}
 	for _, wu := range witnessURL {
 		u, err := url.Parse(wu)
 		if err != nil {
 			klog.Exitf("Invalid witness URL %q: %v", wu, err)
 		}
-		bc := &loggingClient{
+		lc := loggingClient{
 			witness: w_http.NewWitness(u, httpClient),
 			url:     wu,
 		}
-		for f, err := range cfg.Feeders(ctx) {
-			if err != nil {
-				klog.Exitf("Failed to enumerate feedable logs: %v", err)
-			}
-			if r.Match([]byte(f.Log.Origin)) {
-				eg.Go(func() error {
-					if err := rl.Wait(ctx); err != nil {
-						return err
-					}
-					return f.Feeder.FeedFunc()(ctx, f.Log, bc.Update, httpClient, *loopInterval)
-				})
-			}
-		}
+		witnesses = append(witnesses, lc.Update)
+	}
+
+	rOpts := omniwitness.RunFeedOpts{
+		Witnesses:     witnesses,
+		HTTPClient:    httpClient,
+		MaxWitnessQPS: *rateLimit,
+		MatchLogs:     *feed,
+		LogConfig:     cfg,
 	}
-	if err := eg.Wait(); err != nil {
+	if err := omniwitness.RunFeeders(ctx, rOpts); err != nil {
 		klog.Errorf("%v", err)
 	}
 }
@@ -145,7 +133,7 @@ func (lc *loggingClient) Update(ctx context.Context, oldSize uint64, newCP []byt
 		klog.Infof("❌ %s ← %s: %v", lc.url, name, err)
 	}
 
-	return nil, 0, nil
+	return rb, size, err
 }
 
 // multiStringFlag allows a flag to be specified multiple times on the command
@@ -160,3 +148,24 @@ func (ms *multiStringFlag) Set(w string) error {
 	*ms = append(*ms, w)
 	return nil
 }
+
+func httpClientFromFlags() *http.Client {
+	t := &http.Transport{
+		Dial: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).Dial,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ResponseHeaderTimeout: 10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+		MaxIdleConns:          len(witnessURL) + 10,
+		MaxIdleConnsPerHost:   2,
+	}
+	if *httpsInsecure {
+		t.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+	}
+
+	return &http.Client{
+		Transport: t,
+	}
+}

cmd/gcp/omniwitness/main.go

@@ -40,7 +40,6 @@ var (
 	httpTimeout                = flag.Duration("http_timeout", 10*time.Second, "HTTP timeout for outbound requests.")
 
 	pollInterval      = flag.Duration("poll_interval", 1*time.Minute, "Time to wait between polling logs for new checkpoints. Set to 0 to disable polling logs.")
-	feederConcurrency = flag.Uint("feeder_concurrency", 1, "Maximum number of concurrent feeder tasks")
 	additionalLogYaml = flag.String("additional_logs", "", "The path to an optional addition logs YAML file. Entries in this file will be *added* to the logs configured by default")
 )
 
@@ -92,12 +91,11 @@ func main() {
 	}
 
 	opConfig := omniwitness.OperatorConfig{
-		WitnessKeys:      []note.Signer{signer},
-		WitnessVerifier:  signer.Verifier(),
-		FeedInterval:     *pollInterval,
-		NumFeederWorkers: *feederConcurrency,
-		ServeMux:         mux,
-		Logs:             p,
+		WitnessKeys:     []note.Signer{signer},
+		WitnessVerifier: signer.Verifier(),
+		FeedInterval:    *pollInterval,
+		ServeMux:        mux,
+		Logs:            p,
 	}
 	if err := omniwitness.Main(ctx, opConfig, p, httpListener, httpClient); err != nil {
 		klog.Exitf("Main failed: %v", err)

cmd/omniwitness/monolith.go

@@ -55,8 +55,7 @@ var (
 	rateLimit              = flag.Float64("rate_limit", 0, "Maximum number of update requests per second to serve, or zero to disable")
 	httpTimeout            = flag.Duration("http_timeout", 10*time.Second, "HTTP timeout for outbound requests")
 
-	pollInterval      = flag.Duration("poll_interval", 1*time.Minute, "Time to wait between polling logs for new checkpoints. Set to 0 to disable polling logs.")
-	feederConcurrency = flag.Uint("feeder_concurrency", 1, "Maximum number of concurrent feeder tasks")
+	pollInterval = flag.Duration("poll_interval", 1*time.Minute, "Time to wait between polling logs for new checkpoints. Set to 0 to disable polling logs.")
 )
 
 func main() {
@@ -119,7 +118,6 @@ func main() {
 		BastionKey:             bastionKey,
 		RateLimit:              *rateLimit,
 		FeedInterval:           *pollInterval,
-		NumFeederWorkers:       *feederConcurrency,
 	}
 	var p persistence.LogStatePersistence
 	if len(*dbFile) > 0 {

internal/feeder/feeder.go

@@ -19,7 +19,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"time"
 
 	"github.com/cenkalti/backoff/v5"
 	"github.com/transparency-dev/formats/log"
@@ -62,113 +61,71 @@ type FeedOpts struct {
 	Update UpdateFn
 }
 
-// FeedOnce sends the provided checkpoint to the configured witness.
-// This method will block until a witness signature is obtained,
-// or the context becomes done.
-func FeedOnce(ctx context.Context, opts FeedOpts) ([]byte, error) {
-	f := feeder{
-		opts: opts,
-	}
-	return f.feedOnce(ctx)
-}
-
-// Run periodically initiates a feed cycle, fetching a checkpoint from the source log and
-// submitting it to the witness.
-// Calling this function will block until the context is done.
-func Run(ctx context.Context, interval time.Duration, opts FeedOpts) error {
-	f := feeder{
-		opts: opts,
-	}
-	t := time.NewTicker(interval)
-	defer t.Stop()
-	for {
-		// Create a scope with a bounded context so we don't get wedged if something goes wrong.
-		func() {
-			ctx, cancel := context.WithTimeout(ctx, interval)
-			defer cancel()
-
-			if _, err := f.feedOnce(ctx); err != nil {
-				klog.Errorf("Feeding log %q failed: %v", opts.LogSigVerifier.Name(), err)
-			}
-		}()
-
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-t.C:
-		}
-	}
-}
-
-type feeder struct {
-	opts    FeedOpts
-	oldSize uint64
-}
-
-func (f *feeder) feedOnce(ctx context.Context) ([]byte, error) {
-	cp, err := f.opts.FetchCheckpoint(ctx)
+// FeedOnce completes one feeding operation for the log and witness in the provided configuration.
+// The provided sizeHint is size of the log that the caller believes is current on the target witness.
+//
+// Returns a new hint on what the current size of the log on the target witness.
+func FeedOnce(ctx context.Context, sizeHint uint64, opts FeedOpts) (uint64, error) {
+	cp, err := opts.FetchCheckpoint(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("failed to read input checkpoint: %v", err)
+		return sizeHint, fmt.Errorf("failed to read input checkpoint: %v", err)
 	}
 
 	klog.V(2).Infof("CP to feed:\n%s", string(cp))
 
-	cpSubmit, _, _, err := log.ParseCheckpoint(cp, f.opts.LogOrigin, f.opts.LogSigVerifier)
+	cpSubmit, _, _, err := log.ParseCheckpoint(cp, opts.LogOrigin, opts.LogSigVerifier)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse checkpoint: %v", err)
+		return sizeHint, fmt.Errorf("failed to parse checkpoint: %v", err)
 	}
 
-	wCP, err := f.submitToWitness(ctx, cp, *cpSubmit, f.opts)
+	newSize, err := submitToWitness(ctx, sizeHint, cp, *cpSubmit, opts)
 	if err != nil {
-		return nil, fmt.Errorf("witness submission failed: %v", err)
+		return newSize, fmt.Errorf("witness submission failed: %w", err)
 	}
-	return wCP, nil
+	return newSize, nil
 }
 
-// submitToWitness will keep trying to submit the checkpoint to the witness until the context is done.
-func (f *feeder) submitToWitness(ctx context.Context, cpRaw []byte, cpSubmit log.Checkpoint, opts FeedOpts) ([]byte, error) {
-
+// submitToWitness will submit the checkpoint to the witness, retrying up to 3 times if the local checkpoint is stale.
+func submitToWitness(ctx context.Context, sizeHint uint64, cpRaw []byte, cpSubmit log.Checkpoint, opts FeedOpts) (uint64, error) {
 	// Since this func will be executed by the backoff mechanism below, we'll
 	// log any error messages directly in here before returning the error, as
 	// the backoff util doesn't seem to log them itself.
-	submitOp := func() ([]byte, error) {
+	submitOp := func() (uint64, error) {
 		var err error
 		var conP [][]byte
-		if f.oldSize > cpSubmit.Size {
-			return nil, backoff.Permanent(fmt.Errorf("witness checkpoint size (%d) > submit checkpoint size (%d)", f.oldSize, cpSubmit.Size))
+		if sizeHint > cpSubmit.Size {
+			return sizeHint, backoff.Permanent(fmt.Errorf("witness checkpoint size (%d) > submit checkpoint size (%d)", sizeHint, cpSubmit.Size))
 		}
 
 		// The witness may be configured to expect a compact-range type proof, so we need to always
 		// try to build one, even if the witness doesn't have a "latest" checkpoint for this log.
-		conP, err = opts.FetchProof(ctx, f.oldSize, cpSubmit)
+		conP, err = opts.FetchProof(ctx, sizeHint, cpSubmit)
 		if err != nil {
-			e := fmt.Errorf("failed to fetch consistency proof: %v", err)
-			klog.Warning(e.Error())
-			return nil, e
+			e := fmt.Errorf("failed to fetch consistency proof: %w", err)
+			return sizeHint, backoff.Permanent(e)
 		}
-		klog.V(2).Infof("%q: Fetched proof %d -> %d: %x", cpSubmit.Origin, f.oldSize, cpSubmit.Size, conP)
+		klog.V(2).Infof("%q: Fetched proof %d -> %d: %x", cpSubmit.Origin, sizeHint, cpSubmit.Size, conP)
 
-		witnessCp, actualSize, err := opts.Update(ctx, f.oldSize, cpRaw, conP)
+		_, actualSize, err := opts.Update(ctx, sizeHint, cpRaw, conP)
 		switch {
 		case errors.Is(err, witness.ErrCheckpointStale):
-			klog.V(2).Infof("%q: %d is stale, bumping to %d: %x", cpSubmit.Origin, f.oldSize, cpSubmit.Size, conP)
-			f.oldSize = actualSize
-			return nil, backoff.RetryAfter(1)
+			klog.V(2).Infof("%q: %d is stale, bumping to %d: %x", cpSubmit.Origin, sizeHint, cpSubmit.Size, conP)
+			sizeHint = actualSize
+			return sizeHint, backoff.RetryAfter(1)
 		case err != nil:
-			e := fmt.Errorf("%q: failed to submit checkpoint to witness: %v", cpSubmit.Origin, err)
-			klog.Warning(e.Error())
-			return nil, e
+			e := fmt.Errorf("%q: failed to submit checkpoint to witness: %w", cpSubmit.Origin, err)
+			return sizeHint, backoff.Permanent(e)
 		default:
-			if f.oldSize == cpSubmit.Size {
+			if sizeHint == cpSubmit.Size {
 				klog.V(1).Infof("%q: Refreshed witness - @%d: %x", cpSubmit.Origin, cpSubmit.Size, cpSubmit.Hash)
 
 			} else {
-				klog.V(1).Infof("%q: Updated witness - @%d → @%d: %x", cpSubmit.Origin, f.oldSize, cpSubmit.Size, cpSubmit.Hash)
+				klog.V(1).Infof("%q: Updated witness - @%d → @%d: %x", cpSubmit.Origin, sizeHint, cpSubmit.Size, cpSubmit.Hash)
 			}
-			f.oldSize = cpSubmit.Size
+			sizeHint = cpSubmit.Size
 		}
-		return witnessCp, nil
+		return sizeHint, nil
 	}
 
-	return backoff.Retry(ctx, submitOp, backoff.WithBackOff(backoff.NewExponentialBackOff()))
+	return backoff.Retry(ctx, submitOp, backoff.WithBackOff(backoff.NewExponentialBackOff()), backoff.WithMaxTries(3))
 }

internal/feeder/feeder_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/transparency-dev/merkle/rfc6962"
 	sclient "github.com/transparency-dev/serverless-log/client"
 	"github.com/transparency-dev/serverless-log/testdata"
+	"github.com/transparency-dev/witness/internal/witness"
 	"golang.org/x/mod/sumdb/note"
 )
 
@@ -48,7 +49,7 @@ func TestFeedOnce(t *testing.T) {
 				fakeWitness: &fakeWitness{
 					latestCP: testdata.Checkpoint(t, 1),
 				},
-				times: 3,
+				times: 2,
 			}).Update,
 		}, {
 			desc:     "works - TOFU feed",
@@ -88,7 +89,7 @@ func TestFeedOnce(t *testing.T) {
 			Update:          test.update,
 		}
 		t.Run(test.desc, func(t *testing.T) {
-			_, err := FeedOnce(ctx, opts)
+			_, err := FeedOnce(ctx, 0, opts)
 			gotErr := err != nil
 			if test.wantErr != gotErr {
 				t.Fatalf("Got err %v, want err %t", err, test.wantErr)
@@ -105,7 +106,7 @@ type slowWitness struct {
 func (sw *slowWitness) Update(_ context.Context, oldSize uint64, newCP []byte, proof [][]byte) ([]byte, uint64, error) {
 	if sw.times > 0 {
 		sw.times = sw.times - 1
-		return nil, 0, fmt.Errorf("will fail for %d more calls", sw.times)
+		return nil, oldSize, fmt.Errorf("will fail for %d more calls (%w)", sw.times, witness.ErrCheckpointStale)
 	}
 	sw.latestCP = newCP