Repository: ossf/package-feeds (go)

Issue:
Replace lastPoll with a more robust way to track progress.
Currently when a feed is polled, the `FeedGroup.lastPoll` time is used as an argument to `feed.Latest` to specify the cutoff.

The value for `lastPoll` is updated after every feed in a feed group has been polled for latest packages.

This behavior has some serious drawbacks:
- if the FeedGroup has multiple feeds there will be a time gap between when the first feed is completed and when `lastPoll` is updated where updates may occur. This means that updates in the overlap may not be observed.
- the items in the feed may be delayed - e.g. all the items could be 5 minutes behind "now". This means that the cut off may cause updates to be ignored because the timestamp on the items is older than `lastPoll`.

A better approach is:
1. remove `FeedGroup.lastPoll` as an argument to `feed.Latest`.
2. have each feed define its own orderable approach to applying a cutoff.
3. the cutoff should be extracted from the feed itself (i.e. max value observed) and stored with the feed.



Pull request:
Allow ScheduledFeed.Latest to return a cutoff for future invocations
This fixes #382 by ensuring each feed uses a cutoff value that is based on the data observed in the feed.

Patch:
diff --git a/.golangci.yml b/.golangci.yml
index c61e124..de24d4c 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -8,7 +8,7 @@ linters:
     - asciicheck
     - bodyclose
     - deadcode
-    - depguard
+    #- depguard
     - dogsled
     - dupl
     - errcheck
diff --git a/Makefile b/Makefile
index f4ffee7..f56a3aa 100644
--- a/Makefile
+++ b/Makefile
@@ -1,7 +1,7 @@
 GO ?= go
 BIN := bin
 GOOS ?= $(shell uname | tr A-Z a-z)
-GOLANGCI_LINT_VERSION = v1.51.2
+GOLANGCI_LINT_VERSION = v1.53.3
 PROJECT := package-feeds
 
 .PHONY: help
diff --git a/pkg/feeds/crates/crates.go b/pkg/feeds/crates/crates.go
index 4492bbd..175989e 100644
--- a/pkg/feeds/crates/crates.go
+++ b/pkg/feeds/crates/crates.go
@@ -81,11 +81,11 @@ func New(feedOptions feeds.FeedOptions, eventHandler *events.Handler) (*Feed, er
 	}, nil
 }
 
-func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	pkgs := []*feeds.Package{}
 	packages, err := fetchPackages(feed.baseURL)
 	if err != nil {
-		return pkgs, []error{err}
+		return pkgs, cutoff, []error{err}
 	}
 	for _, pkg := range packages {
 		pkg := feeds.NewPackage(pkg.UpdatedAt, pkg.Name, pkg.NewestVersion, FeedName)
@@ -93,8 +93,9 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 	}
 	feed.lossyFeedAlerter.ProcessPackages(FeedName, pkgs)
 
+	newCutoff := feeds.FindCutoff(cutoff, pkgs)
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
-	return pkgs, []error{}
+	return pkgs, newCutoff, []error{}
 }
 
 func (feed Feed) GetName() string {
diff --git a/pkg/feeds/feed.go b/pkg/feeds/feed.go
index 3018e1c..690d59e 100644
--- a/pkg/feeds/feed.go
+++ b/pkg/feeds/feed.go
@@ -16,7 +16,7 @@ type UnsupportedOptionError struct {
 }
 
 type ScheduledFeed interface {
-	Latest(cutoff time.Time) ([]*Package, []error)
+	Latest(cutoff time.Time) ([]*Package, time.Time, []error)
 	GetFeedOptions() FeedOptions
 	GetName() string
 }
@@ -70,13 +70,22 @@ func NewArtifact(created time.Time, name, version, artifactID, feed string) *Pac
 func ApplyCutoff(pkgs []*Package, cutoff time.Time) []*Package {
 	filteredPackages := []*Package{}
 	for _, pkg := range pkgs {
-		if !pkg.CreatedDate.Before(cutoff) {
+		if pkg.CreatedDate.After(cutoff) {
 			filteredPackages = append(filteredPackages, pkg)
 		}
 	}
 	return filteredPackages
 }
 
+func FindCutoff(cutoff time.Time, pkgs []*Package) time.Time {
+	for _, pkg := range pkgs {
+		if pkg.CreatedDate.After(cutoff) {
+			cutoff = pkg.CreatedDate
+		}
+	}
+	return cutoff
+}
+
 func (err UnsupportedOptionError) Error() string {
 	return fmt.Sprintf("unsupported option `%v` supplied to %v feed", err.Option, err.Feed)
 }
diff --git a/pkg/feeds/goproxy/goproxy.go b/pkg/feeds/goproxy/goproxy.go
index 6765eb4..cc5c720 100644
--- a/pkg/feeds/goproxy/goproxy.go
+++ b/pkg/feeds/goproxy/goproxy.go
@@ -100,18 +100,19 @@ func New(feedOptions feeds.FeedOptions) (*Feed, error) {
 	}, nil
 }
 
-func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	pkgs := []*feeds.Package{}
 	packages, err := fetchPackages(feed.baseURL, cutoff)
 	if err != nil {
-		return pkgs, []error{err}
+		return pkgs, cutoff, []error{err}
 	}
 	for _, pkg := range packages {
 		pkg := feeds.NewPackage(pkg.ModifiedDate, pkg.Title, pkg.Version, FeedName)
 		pkgs = append(pkgs, pkg)
 	}
+	newCutoff := feeds.FindCutoff(cutoff, pkgs)
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
-	return pkgs, []error{}
+	return pkgs, newCutoff, []error{}
 }
 
 func (feed Feed) GetName() string {
diff --git a/pkg/feeds/npm/npm.go b/pkg/feeds/npm/npm.go
index 6577eae..569ab16 100644
--- a/pkg/feeds/npm/npm.go
+++ b/pkg/feeds/npm/npm.go
@@ -390,7 +390,7 @@ func New(feedOptions feeds.FeedOptions, eventHandler *events.Handler) (*Feed, er
 	}, nil
 }
 
-func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	var pkgs []*feeds.Package
 	var errs []error
 
@@ -402,7 +402,7 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 
 	if len(pkgs) == 0 {
 		// If none of the packages were successfully polled for, return early.
-		return nil, append(errs, feeds.ErrNoPackagesPolled)
+		return nil, cutoff, append(errs, feeds.ErrNoPackagesPolled)
 	}
 
 	// Ensure packages are sorted by CreatedDate in order of most recent, as goroutine
@@ -411,6 +411,9 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 		return pkgs[j].CreatedDate.Before(pkgs[i].CreatedDate)
 	})
 
+	// After sorting the first entry should be the largest.
+	newCutoff := pkgs[0].CreatedDate
+
 	// TODO: Add an event for checking if the previous package list contains entries
 	// that do not exist in the latest package list when polling for critical packages.
 	// This can highlight cases where specific versions have been unpublished.
@@ -419,7 +422,7 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 	}
 
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
-	return pkgs, errs
+	return pkgs, newCutoff, errs
 }
 
 func (feed Feed) GetName() string {
diff --git a/pkg/feeds/nuget/nuget.go b/pkg/feeds/nuget/nuget.go
index b192c1f..787b942 100644
--- a/pkg/feeds/nuget/nuget.go
+++ b/pkg/feeds/nuget/nuget.go
@@ -176,18 +176,18 @@ func New(feedOptions feeds.FeedOptions) (*Feed, error) {
 // Latest will parse all creation events for packages in the nuget.org catalog feed
 // for packages that have been published since the cutoff
 // https://docs.microsoft.com/en-us/nuget/api/catalog-resource
-func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	pkgs := []*feeds.Package{}
 	var errs []error
 
 	catalogService, err := fetchCatalogService(feed.baseURL)
 	if err != nil {
-		return nil, append(errs, err)
+		return nil, cutoff, append(errs, err)
 	}
 
 	catalogPages, err := fetchCatalogPages(catalogService.URI)
 	if err != nil {
-		return nil, append(errs, err)
+		return nil, cutoff, append(errs, err)
 	}
 
 	for _, catalogPage := range catalogPages {
@@ -220,9 +220,10 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 			pkgs = append(pkgs, pkg)
 		}
 	}
+	newCutoff := feeds.FindCutoff(cutoff, pkgs)
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
 
-	return pkgs, errs
+	return pkgs, newCutoff, errs
 }
 
 func (feed Feed) GetName() string {
diff --git a/pkg/feeds/packagist/packagist.go b/pkg/feeds/packagist/packagist.go
index 9a1e29f..9a664ce 100644
--- a/pkg/feeds/packagist/packagist.go
+++ b/pkg/feeds/packagist/packagist.go
@@ -115,12 +115,12 @@ func fetchVersionInformation(versionHost string, action actions) ([]*feeds.Packa
 }
 
 // Latest returns all package updates of packagist packages since cutoff.
-func (f Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (f Feed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	pkgs := []*feeds.Package{}
 	var errs []error
 	packages, err := fetchPackages(f.updateHost, cutoff)
 	if err != nil {
-		return nil, append(errs, err)
+		return nil, cutoff, append(errs, err)
 	}
 	for _, pkg := range packages {
 		if time.Unix(pkg.Time, 0).Before(cutoff) {
@@ -136,8 +136,9 @@ func (f Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 		}
 		pkgs = append(pkgs, updates...)
 	}
+	newCutoff := feeds.FindCutoff(cutoff, pkgs)
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
-	return pkgs, errs
+	return pkgs, newCutoff, errs
 }
 
 func (f Feed) GetName() string {
diff --git a/pkg/feeds/pypi/pypi.go b/pkg/feeds/pypi/pypi.go
index c35c513..e6dcd10 100644
--- a/pkg/feeds/pypi/pypi.go
+++ b/pkg/feeds/pypi/pypi.go
@@ -166,7 +166,7 @@ func New(feedOptions feeds.FeedOptions, eventHandler *events.Handler) (*Feed, er
 	}, nil
 }
 
-func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	pkgs := []*feeds.Package{}
 	var pypiPackages []*Package
 	var errs []error
@@ -178,14 +178,14 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 		// data.
 		pypiPackages, err = fetchPackages(feed.baseURL)
 		if err != nil {
-			return nil, append(errs, err)
+			return nil, cutoff, append(errs, err)
 		}
 	} else {
 		// Fetch specific packages individually from configured packages list.
 		pypiPackages, errs = fetchCriticalPackages(feed.baseURL, *feed.packages)
 		if len(pypiPackages) == 0 {
 			// If none of the packages were successfully polled for, return early.
-			return nil, append(errs, feeds.ErrNoPackagesPolled)
+			return nil, cutoff, append(errs, feeds.ErrNoPackagesPolled)
 		}
 	}
 
@@ -209,8 +209,9 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 		feed.lossyFeedAlerter.ProcessPackages(FeedName, pkgs)
 	}
 
+	newCutoff := feeds.FindCutoff(cutoff, pkgs)
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
-	return pkgs, errs
+	return pkgs, newCutoff, errs
 }
 
 func (feed Feed) GetPackageList() *[]string {
diff --git a/pkg/feeds/pypi/pypi_artifacts.go b/pkg/feeds/pypi/pypi_artifacts.go
index 79ea050..55484cf 100644
--- a/pkg/feeds/pypi/pypi_artifacts.go
+++ b/pkg/feeds/pypi/pypi_artifacts.go
@@ -29,18 +29,19 @@ func NewArtifactFeed(feedOptions feeds.FeedOptions) (*ArtifactFeed, error) {
 	}, nil
 }
 
-func (feed ArtifactFeed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (feed ArtifactFeed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	client, err := xmlrpc.NewClient(feed.baseURL, nil)
 	if err != nil {
-		return nil, []error{err}
+		return nil, cutoff, []error{err}
 	}
 
 	changelogEntries, err := getPyPIChangeLog(client, cutoff)
 	if err != nil {
-		return nil, []error{err}
+		return nil, cutoff, []error{err}
 	}
 
-	return getUploadedArtifacts(changelogEntries), nil
+	pkgs := getUploadedArtifacts(changelogEntries)
+	return pkgs, feeds.FindCutoff(cutoff, pkgs), nil
 }
 
 func (feed ArtifactFeed) GetFeedOptions() feeds.FeedOptions {
diff --git a/pkg/feeds/rubygems/rubygems.go b/pkg/feeds/rubygems/rubygems.go
index e680973..26165e7 100644
--- a/pkg/feeds/rubygems/rubygems.go
+++ b/pkg/feeds/rubygems/rubygems.go
@@ -64,7 +64,7 @@ func New(feedOptions feeds.FeedOptions, eventHandler *events.Handler) (*Feed, er
 	}, nil
 }
 
-func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
 	pkgs := []*feeds.Package{}
 	packages := make(map[string]*Package)
 	var errs []error
@@ -72,7 +72,7 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 	newPackagesURL, err := url.JoinPath(feed.baseURL, activityPath, "latest.json")
 	if err != nil {
 		// Failure to construct a url should lead to a hard failure.
-		return nil, append(errs, err)
+		return nil, cutoff, append(errs, err)
 	}
 	newPackages, err := fetchPackages(newPackagesURL)
 	if err != nil {
@@ -86,7 +86,7 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 	updatedPackagesURL, err := url.JoinPath(feed.baseURL, activityPath, "just_updated.json")
 	if err != nil {
 		// Failure to construct a url should lead to a hard failure.
-		return nil, append(errs, err)
+		return nil, cutoff, append(errs, err)
 	}
 	updatedPackages, err := fetchPackages(updatedPackagesURL)
 	if err != nil {
@@ -104,8 +104,9 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 	}
 	feed.lossyFeedAlerter.ProcessPackages(FeedName, pkgs)
 
+	newCutoff := feeds.FindCutoff(cutoff, pkgs)
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
-	return pkgs, errs
+	return pkgs, newCutoff, errs
 }
 
 func (feed Feed) GetName() string {
diff --git a/pkg/scheduler/feed_group.go b/pkg/scheduler/feed_group.go
index 8e39d36..5ad97a0 100644
--- a/pkg/scheduler/feed_group.go
+++ b/pkg/scheduler/feed_group.go
@@ -17,10 +17,15 @@ var (
 	errPub  = errors.New("error when publishing packages")
 )
 
+type feedEntry struct {
+	feed     feeds.ScheduledFeed
+	lastPoll time.Time
+}
+
 type FeedGroup struct {
-	feeds     []feeds.ScheduledFeed
-	publisher publisher.Publisher
-	lastPoll  time.Time
+	feeds         []*feedEntry
+	publisher     publisher.Publisher
+	initialCutoff time.Time
 }
 
 type groupResult struct {
@@ -31,15 +36,22 @@ type groupResult struct {
 
 //nolint:lll
 func NewFeedGroup(scheduledFeeds []feeds.ScheduledFeed, pub publisher.Publisher, initialCutoff time.Duration) *FeedGroup {
-	return &FeedGroup{
-		feeds:     scheduledFeeds,
-		publisher: pub,
-		lastPoll:  time.Now().UTC().Add(-initialCutoff),
+	fg := &FeedGroup{
+		publisher:     pub,
+		initialCutoff: time.Now().UTC().Add(-initialCutoff),
+		feeds:         make([]*feedEntry, 0),
+	}
+	for _, feed := range scheduledFeeds {
+		fg.AddFeed(feed)
 	}
+	return fg
 }
 
 func (fg *FeedGroup) AddFeed(feed feeds.ScheduledFeed) {
-	fg.feeds = append(fg.feeds, feed)
+	fg.feeds = append(fg.feeds, &feedEntry{
+		feed:     feed,
+		lastPoll: fg.initialCutoff,
+	})
 }
 
 func (fg *FeedGroup) Run() {
@@ -71,15 +83,15 @@ func (fg *FeedGroup) pollAndPublish() groupResult {
 // Poll fetches the latest packages from each registered feed.
 func (fg *FeedGroup) poll() ([]*feeds.Package, error) {
 	results := make(chan pollResult, len(fg.feeds))
-	for _, feed := range fg.feeds {
-		go func(feed feeds.ScheduledFeed) {
+	for _, f := range fg.feeds {
+		go func(f *feedEntry) {
 			result := pollResult{
-				name: feed.GetName(),
-				feed: feed,
+				name: f.feed.GetName(),
+				feed: f.feed,
 			}
-			result.packages, result.errs = feed.Latest(fg.lastPoll)
+			result.packages, f.lastPoll, result.errs = f.feed.Latest(f.lastPoll)
 			results <- result
-		}(feed)
+		}(f)
 	}
 	errs := []error{}
 	packages := []*feeds.Package{}
@@ -105,9 +117,8 @@ func (fg *FeedGroup) poll() ([]*feeds.Package, error) {
 	if len(errs) == 0 {
 		err = nil
 	}
-	fg.lastPoll = time.Now().UTC()
 
-	log.Printf("%d packages processed", len(packages))
+	log.WithField("time", time.Now().UTC()).Printf("%d packages processed", len(packages))
 	return packages, err
 }
 
diff --git a/pkg/scheduler/mocks.go b/pkg/scheduler/mocks.go
index 20972ee..3392e1e 100644
--- a/pkg/scheduler/mocks.go
+++ b/pkg/scheduler/mocks.go
@@ -1,3 +1,4 @@
+//nolint:gocritic
 package scheduler
 
 import (
@@ -10,6 +11,7 @@ import (
 type mockFeed struct {
 	packages []*feeds.Package
 	errs     []error
+	cutoff   time.Time
 	options  feeds.FeedOptions
 }
 
@@ -21,8 +23,8 @@ func (feed mockFeed) GetFeedOptions() feeds.FeedOptions {
 	return feed.options
 }
 
-func (feed mockFeed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
-	return feed.packages, feed.errs
+func (feed mockFeed) Latest(cutoff time.Time) ([]*feeds.Package, time.Time, []error) {
+	return feed.packages, feed.cutoff, feed.errs
 }
 
 type mockPublisher struct {
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 279841c..780848c 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -58,7 +58,7 @@ func (s *Scheduler) Run(initialCutoff time.Duration, enableDefaultTimer bool) er
 	for schedule, feedGroup := range schedules {
 		var feedNames []string
 		for _, f := range feedGroup.feeds {
-			feedNames = append(feedNames, f.GetName())
+			feedNames = append(feedNames, f.feed.GetName())
 		}
 
 		if schedule == "" {


Test patch:
diff --git a/pkg/feeds/crates/crates_test.go b/pkg/feeds/crates/crates_test.go
index 3657dc1..971cfae 100644
--- a/pkg/feeds/crates/crates_test.go
+++ b/pkg/feeds/crates/crates_test.go
@@ -27,11 +27,17 @@ func TestCratesLatest(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("feed.Latest returned error: %v", err)
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2021, 3, 19, 13, 36, 33, 0, time.UTC)
+	if gotCutoff.Sub(wantCutoff).Abs() > time.Second {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
+
 	if pkgs[0].Name != "FooPackage" {
 		t.Errorf("Unexpected package `%s` found in place of expected `FooPackage`", pkgs[0].Name)
 	}
@@ -67,7 +73,10 @@ func TestCratesNotFound(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	_, errs := feed.Latest(cutoff)
+	_, gotCutoff, errs := feed.Latest(cutoff)
+	if cutoff != gotCutoff {
+		t.Error("feed.Latest() cutoff should be unchanged if an error is returned")
+	}
 	if len(errs) == 0 {
 		t.Fatalf("feed.Latest() was successful when an error was expected")
 	}
diff --git a/pkg/feeds/goproxy/goproxy_test.go b/pkg/feeds/goproxy/goproxy_test.go
index b9eb8ec..0e77a48 100644
--- a/pkg/feeds/goproxy/goproxy_test.go
+++ b/pkg/feeds/goproxy/goproxy_test.go
@@ -26,11 +26,16 @@ func TestGoProxyLatest(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("feed.Latest returned error: %v", err)
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2019, 4, 10, 20, 30, 2, 0, time.UTC)
+	if gotCutoff.Sub(wantCutoff).Abs() > time.Second {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
 	if pkgs[0].Name != "golang.org/x/foo" {
 		t.Errorf("Unexpected package `%s` found in place of expected `golang.org/x/foo`", pkgs[0].Name)
 	}
@@ -66,7 +71,10 @@ func TestGoproxyNotFound(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	_, errs := feed.Latest(cutoff)
+	_, gotCutoff, errs := feed.Latest(cutoff)
+	if cutoff != gotCutoff {
+		t.Error("feed.Latest() cutoff should be unchanged if an error is returned")
+	}
 	if len(errs) == 0 {
 		t.Fatalf("feed.Latest() was successful when an error was expected")
 	}
diff --git a/pkg/feeds/npm/npm_test.go b/pkg/feeds/npm/npm_test.go
index 307c5b6..9c41adb 100644
--- a/pkg/feeds/npm/npm_test.go
+++ b/pkg/feeds/npm/npm_test.go
@@ -35,11 +35,17 @@ func TestNpmLatest(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("feed.Latest returned error: %v", errs[len(errs)-1])
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2021, 5, 11, 18, 32, 1, 0, time.UTC)
+	if gotCutoff != wantCutoff {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
+
 	if pkgs[0].Name != "FooPackage" {
 		t.Errorf("Unexpected package `%s` found in place of expected `FooPackage`", pkgs[0].Name)
 	}
@@ -136,7 +142,7 @@ func TestNpmCritical(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("Failed to call Latest() with err: %v", errs[len(errs)-1])
 	}
@@ -145,6 +151,12 @@ func TestNpmCritical(t *testing.T) {
 		t.Fatalf("Latest() produced %v packages instead of the expected 7", len(pkgs))
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2021, 5, 11, 18, 32, 1, 0, time.UTC)
+	if gotCutoff != wantCutoff {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
+
 	pkgMap := map[string]map[string]*feeds.Package{}
 	pkgMap["FooPackage"] = map[string]*feeds.Package{}
 	pkgMap["BarPackage"] = map[string]*feeds.Package{}
@@ -192,7 +204,7 @@ func TestNpmCriticalUnpublished(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, _, errs := feed.Latest(cutoff)
 
 	if len(errs) != 1 {
 		t.Fatalf("feed.Latest() returned %v errors when 1 was expected", len(errs))
@@ -274,7 +286,10 @@ func TestNpmNotFound(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	_, errs := feed.Latest(cutoff)
+	_, gotCutoff, errs := feed.Latest(cutoff)
+	if cutoff != gotCutoff {
+		t.Error("feed.Latest() cutoff should be unchanged if an error is returned")
+	}
 	if len(errs) != 2 {
 		t.Fatalf("feed.Latest() returned %v errors when 2 were expected", len(errs))
 	}
@@ -304,7 +319,7 @@ func TestNpmPartialNotFound(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, _, errs := feed.Latest(cutoff)
 	if len(errs) != 1 {
 		t.Fatalf("feed.Latest() returned %v errors when 1 was expected", len(errs))
 	}
@@ -343,7 +358,7 @@ func TestNpmCriticalPartialNotFound(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, _, errs := feed.Latest(cutoff)
 	if len(errs) != 1 {
 		t.Fatalf("feed.Latest() returned %v errors when 1 was expected", len(errs))
 	}
diff --git a/pkg/feeds/nuget/nuget_test.go b/pkg/feeds/nuget/nuget_test.go
index 28b7ae9..6c729b5 100644
--- a/pkg/feeds/nuget/nuget_test.go
+++ b/pkg/feeds/nuget/nuget_test.go
@@ -39,11 +39,19 @@ func TestCanParseFeed(t *testing.T) {
 
 	cutoff := time.Now().Add(-5 * time.Minute)
 
-	results, errs := sut.Latest(cutoff)
+	results, gotCutoff, errs := sut.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatal(errs[len(errs)-1])
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	// This time is automatically generated at approx 1 minutes ago. The threshold of 70s
+	// ensures this test doesn't fail with the multiple mocked API requests.
+	wantCutoff := time.Now().UTC()
+	if gotCutoff == cutoff || gotCutoff.Sub(wantCutoff).Abs() > (70*time.Second) {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
+
 	if len(results) != 1 {
 		t.Fatalf("1 result expected but %d retrieved", len(results))
 	}
diff --git a/pkg/feeds/packagist/packagist_test.go b/pkg/feeds/packagist/packagist_test.go
index eecb825..076f1fd 100644
--- a/pkg/feeds/packagist/packagist_test.go
+++ b/pkg/feeds/packagist/packagist_test.go
@@ -29,13 +29,19 @@ func TestFetch(t *testing.T) {
 	feed.versionHost = srv.URL
 
 	cutoff := time.Unix(1614513658, 0)
-	latest, errs := feed.Latest(cutoff)
+	latest, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("got error: %v", errs[len(errs)-1])
 	}
 	if len(latest) == 0 {
 		t.Fatalf("did not get any updates")
 	}
+
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2021, 2, 28, 12, 20, 3, 0, time.UTC)
+	if gotCutoff.Sub(wantCutoff).Abs() > time.Second {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
 	for _, pkg := range latest {
 		if pkg.CreatedDate.Before(cutoff) {
 			t.Fatalf("package returned that was updated before cutoff %f", pkg.CreatedDate.Sub(cutoff).Minutes())
@@ -64,7 +70,10 @@ func TestPackagistNotFound(t *testing.T) {
 	feed.versionHost = srv.URL
 
 	cutoff := time.Unix(1614513658, 0)
-	_, errs := feed.Latest(cutoff)
+	_, gotCutoff, errs := feed.Latest(cutoff)
+	if cutoff != gotCutoff {
+		t.Error("feed.Latest() cutoff should be unchanged if an error is returned")
+	}
 	if len(errs) != 1 {
 		t.Fatalf("feed.Latest() returned %v errors when 1 was expected", len(errs))
 	}
diff --git a/pkg/feeds/pypi/pypi_artifacts_test.go b/pkg/feeds/pypi/pypi_artifacts_test.go
index 3ff8a48..826410d 100644
--- a/pkg/feeds/pypi/pypi_artifacts_test.go
+++ b/pkg/feeds/pypi/pypi_artifacts_test.go
@@ -403,11 +403,19 @@ func TestPyPIArtifactsLive(t *testing.T) {
 	}
 
 	cutoff := time.Now().AddDate(0, 0, -1)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("feed.Latest returned error: %v", err)
 	}
 
+	if cutoff == gotCutoff {
+		t.Errorf("Latest() cutoff did not change")
+	}
+
+	if len(pkgs) == 0 {
+		t.Fatalf("Latest() returned no packages")
+	}
+
 	for _, p := range pkgs {
 		if p.Name == "" {
 			t.Errorf("Package has no name: %s", p)
diff --git a/pkg/feeds/pypi/pypi_test.go b/pkg/feeds/pypi/pypi_test.go
index 40152ed..024c79d 100644
--- a/pkg/feeds/pypi/pypi_test.go
+++ b/pkg/feeds/pypi/pypi_test.go
@@ -27,11 +27,17 @@ func TestPypiLatest(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("feed.Latest returned error: %v", err)
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2021, 3, 19, 12, 1, 4, 0, time.UTC)
+	if gotCutoff.Sub(wantCutoff).Abs() > time.Second {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
+
 	if pkgs[0].Name != "FooPackage" {
 		t.Errorf("Unexpected package `%s` found in place of expected `FooPackage`", pkgs[0].Name)
 	}
@@ -74,11 +80,17 @@ func TestPypiCriticalLatest(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("Failed to call Latest() with err: %v", errs[len(errs)-1])
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2021, 3, 27, 22, 16, 26, 0, time.UTC)
+	if gotCutoff.Sub(wantCutoff).Abs() > time.Second {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
+
 	const expectedNumPackages = 4
 	if len(pkgs) != expectedNumPackages {
 		t.Fatalf("Latest() produced %v packages instead of the expected %v", len(pkgs), expectedNumPackages)
@@ -127,7 +139,10 @@ func TestPypiAllNotFound(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	_, errs := feed.Latest(cutoff)
+	_, gotCutoff, errs := feed.Latest(cutoff)
+	if cutoff != gotCutoff {
+		t.Error("feed.Latest() cutoff should be unchanged if an error is returned")
+	}
 	if len(errs) != 3 {
 		t.Fatalf("feed.Latest() returned %v errors when 3 were expected", len(errs))
 	}
@@ -158,7 +173,7 @@ func TestPypiCriticalPartialNotFound(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, _, errs := feed.Latest(cutoff)
 	if len(errs) != 1 {
 		t.Fatalf("feed.Latest() returned %v errors when 1 was expected", len(errs))
 	}
diff --git a/pkg/feeds/rubygems/rubygems_test.go b/pkg/feeds/rubygems/rubygems_test.go
index 835b147..24adb7a 100644
--- a/pkg/feeds/rubygems/rubygems_test.go
+++ b/pkg/feeds/rubygems/rubygems_test.go
@@ -28,11 +28,17 @@ func TestRubyLatest(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, gotCutoff, errs := feed.Latest(cutoff)
 	if len(errs) != 0 {
 		t.Fatalf("feed.Latest returned error: %v", errs[len(errs)-1])
 	}
 
+	// Returned cutoff should match the newest package creation time of packages retrieved.
+	wantCutoff := time.Date(2021, 3, 19, 13, 0, 43, 0, time.UTC)
+	if gotCutoff.Sub(wantCutoff).Abs() > time.Second {
+		t.Errorf("Latest() cutoff %v, want %v", gotCutoff, wantCutoff)
+	}
+
 	var fooPkg *feeds.Package
 	var barPkg *feeds.Package
 
@@ -78,7 +84,10 @@ func TestRubyGemsNotFound(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	_, errs := feed.Latest(cutoff)
+	_, gotCutoff, errs := feed.Latest(cutoff)
+	if cutoff != gotCutoff {
+		t.Error("feed.Latest() cutoff should be unchanged if an error is returned")
+	}
 	if len(errs) == 0 {
 		t.Fatalf("feed.Latest() was successful when an error was expected")
 	}
@@ -103,7 +112,7 @@ func TestRubyGemsPartialNotFound(t *testing.T) {
 	}
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, errs := feed.Latest(cutoff)
+	pkgs, _, errs := feed.Latest(cutoff)
 	if len(errs) != 1 {
 		t.Fatalf("feed.Latest() returned %v errors when 1 was expected", len(errs))
 	}
diff --git a/pkg/scheduler/feed_group_test.go b/pkg/scheduler/feed_group_test.go
index 836ed0b..cf7aad4 100644
--- a/pkg/scheduler/feed_group_test.go
+++ b/pkg/scheduler/feed_group_test.go
@@ -35,7 +35,6 @@ func TestFeedGroupPoll(t *testing.T) {
 	var pub publisher.Publisher = mockPub
 
 	feedGroup := NewFeedGroup(mockFeeds, pub, time.Minute)
-	startLastPollValue := feedGroup.lastPoll
 
 	pkgs, err := feedGroup.poll()
 	if err != nil {
@@ -44,9 +43,6 @@ func TestFeedGroupPoll(t *testing.T) {
 	if len(pkgs) != 4 {
 		t.Fatalf("poll() returned %v packages when 4 were expected", len(pkgs))
 	}
-	if startLastPollValue.Equal(feedGroup.lastPoll) {
-		t.Fatalf("Feed Group did not update last poll as expected")
-	}
 }
 
 func TestFeedGroupPollWithErr(t *testing.T) {
@@ -68,7 +64,6 @@ func TestFeedGroupPollWithErr(t *testing.T) {
 	var pub publisher.Publisher = mockPub
 
 	feedGroup := NewFeedGroup(mockFeeds, pub, time.Minute)
-	startLastPollValue := feedGroup.lastPoll
 
 	pkgs, err := feedGroup.poll()
 	if err == nil {
@@ -80,9 +75,6 @@ func TestFeedGroupPollWithErr(t *testing.T) {
 	if len(pkgs) != 2 {
 		t.Fatalf("Expected 2 packages alongside errors but found %v", len(pkgs))
 	}
-	if startLastPollValue.Equal(feedGroup.lastPoll) {
-		t.Fatalf("Feed Group did not update last poll as expected")
-	}
 }
 
 func TestFeedGroupPublish(t *testing.T) {