fix(importer): add lookback window and pre-fetch check for REST imports

go/internal/database/datastore/models.go

@@ -162,6 +162,7 @@ type SourceRepository struct {
 	DirectoryPath           string                      `datastore:"directory_path"`
 	LastSyncedHash          string                      `datastore:"last_synced_hash"`
 	LastUpdateDate          *time.Time                  `datastore:"last_update_date"`
+	RESTHTTPLastModified    *time.Time                  `datastore:"rest_http_last_modified,omitempty"`
 	IgnorePatterns          []string                    `datastore:"ignore_patterns"`
 	Editable                bool                        `datastore:"editable"`
 	Extension               string                      `datastore:"extension"`

go/internal/database/datastore/source_repository.go

@@ -124,6 +124,7 @@ func (sr *SourceRepository) toModel() *models.SourceRepository {
 		msr.REST = &models.SourceRepoREST{
 			URL:                     sr.RESTAPIURL,
 			LastUpdated:             sr.LastUpdateDate,
+			HTTPLastModified:        sr.RESTHTTPLastModified,
 			IgnoreLastImportTime:    sr.IgnoreLastImportTime,
 			IgnoreDeletionThreshold: sr.IgnoreDeletionThreshold,
 		}
@@ -173,6 +174,7 @@ func newSourceRepositoryFromModel(r *models.SourceRepository) *SourceRepository
 	if r.REST != nil {
 		sr.RESTAPIURL = r.REST.URL
 		sr.LastUpdateDate = r.REST.LastUpdated
+		sr.RESTHTTPLastModified = r.REST.HTTPLastModified
 		sr.IgnoreLastImportTime = r.REST.IgnoreLastImportTime
 		sr.IgnoreDeletionThreshold = r.REST.IgnoreDeletionThreshold
 	}

go/internal/database/datastore/source_repository_test.go

@@ -106,7 +106,8 @@ func TestSourceRepositoryStore_Update(t *testing.T) {
 				Type: models.SourceRepositoryTypeREST,
 				REST: &models.SourceRepoREST{
 					URL:                     "http://rest.url",
-					LastUpdated:             &testTime2,
+					LastUpdated:             &testTime,
+					HTTPLastModified:        &testTime2,
 					IgnoreLastImportTime:    true,
 					IgnoreDeletionThreshold: true,
 				},
@@ -115,7 +116,8 @@ func TestSourceRepositoryStore_Update(t *testing.T) {
 				Name:                    "test-rest-repo",
 				Type:                    models.SourceRepositoryTypeREST,
 				RESTAPIURL:              "http://rest.url",
-				LastUpdateDate:          &testTime2,
+				LastUpdateDate:          &testTime,
+				RESTHTTPLastModified:    &testTime2,
 				IgnoreLastImportTime:    true,
 				IgnoreDeletionThreshold: true,
 			},
@@ -251,6 +253,29 @@ func TestSourceRepositoryStore_Get(t *testing.T) {
 	if diff := cmp.Diff(bucketRepo, gotBucket, cmpopts.EquateEmpty()); diff != "" {
 		t.Errorf("Get() Bucket repo mismatch (-want +got):\n%s", diff)
 	}
+
+	testTime2 := time.Date(2023, 10, 27, 0, 0, 0, 0, time.UTC)
+	restRepo := &models.SourceRepository{
+		Name: "test-rest-repo",
+		Type: models.SourceRepositoryTypeREST,
+		REST: &models.SourceRepoREST{
+			URL:              "http://rest.url",
+			LastUpdated:      &testTime,
+			HTTPLastModified: &testTime2,
+		},
+		GitAnalysis: &models.GitAnalysisConfig{},
+	}
+	if err := store.Update(ctx, restRepo.Name, restRepo); err != nil {
+		t.Fatalf("Failed to setup REST test data: %v", err)
+	}
+
+	gotREST, err := store.Get(ctx, "test-rest-repo")
+	if err != nil {
+		t.Fatalf("Failed to get REST repo: %v", err)
+	}
+	if diff := cmp.Diff(restRepo, gotREST, cmpopts.EquateEmpty()); diff != "" {
+		t.Errorf("Get() REST repo mismatch (-want +got):\n%s", diff)
+	}
 }
 
 func TestSourceRepositoryStore_All(t *testing.T) {

go/internal/importer/importer.go

@@ -549,6 +549,8 @@ type WorkItem struct {
 
 	SourceRepository       string
 	SourcePath             string
+	KnownID                string
+	KnownModified          time.Time
 	Format                 RecordFormat
 	KeyPath                string
 	Strict                 bool
@@ -605,6 +607,25 @@ func processUpdate(ctx context.Context, config Config, item WorkItem) {
 	sourcePath := item.SourcePath
 	strict := config.StrictValidation && item.Strict
 
+	if !item.IsReimport && item.CompareAgainstDatabase && item.Action == ActionImport &&
+		item.KnownID != "" && !item.KnownModified.IsZero() {
+		dbModified, err := config.VulnerabilityStore.GetSourceModified(ctx, item.KnownID)
+		if err != nil && !errors.Is(err, models.ErrNotFound) {
+			logger.ErrorContext(ctx, "Failed to get source modified",
+				slog.Any("error", err),
+				slog.String("source", sourceRepoName),
+				slog.String("path", sourcePath),
+				slog.String("id", item.KnownID))
+
+			return
+		}
+		// Compare at whole-second granularity so sub-second timestamps in a REST index
+		// do not trigger redundant fetches when the individual record truncates to seconds.
+		if err == nil && !item.KnownModified.Truncate(time.Second).After(dbModified.Truncate(time.Second)) {
+			return
+		}
+	}
+
 	unmarshalOptions := protojson.UnmarshalOptions{
 		DiscardUnknown: !strict,
 	}

go/internal/importer/importer_test.go

@@ -359,3 +359,72 @@ func TestIsReconcileOutdated(t *testing.T) {
 		})
 	}
 }
+
+func TestProcessUpdate_KnownModifiedCheck(t *testing.T) {
+	baseTime := time.Date(2026, 1, 10, 12, 0, 0, 0, time.UTC)
+
+	tests := []struct {
+		name          string
+		dbMod         time.Time
+		knownModified time.Time
+		wantOpened    bool
+		wantPublished bool
+	}{
+		{
+			name:          "skips Open when KnownModified matches dbMod at whole-second precision",
+			dbMod:         baseTime,
+			knownModified: baseTime.Add(861 * time.Millisecond),
+			wantOpened:    false,
+			wantPublished: false,
+		},
+		{
+			name:          "opens and publishes when KnownModified is newer than dbMod",
+			dbMod:         baseTime,
+			knownModified: baseTime.Add(time.Second),
+			wantOpened:    true,
+			wantPublished: true,
+		},
+		{
+			name:          "opens and publishes when missing from database",
+			knownModified: baseTime,
+			wantOpened:    true,
+			wantPublished: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rawMods := map[string]time.Time{}
+			if !tt.dbMod.IsZero() {
+				rawMods["CGA-1234"] = tt.dbMod
+			}
+			pub := &testutils.MockPublisher{}
+			cfg := Config{
+				Publisher:          pub,
+				VulnerabilityStore: &mockVulnerabilityStore{RawMods: rawMods},
+			}
+
+			opened := false
+			processUpdate(t.Context(), cfg, WorkItem{
+				SourceRecord: mockSourceRecord{
+					DataToRead: []byte(`{"id": "CGA-1234", "modified": "2026-01-10T12:00:01Z"}`),
+					OnOpen:     func() { opened = true },
+				},
+				SourceRepository:       "chainguard",
+				SourcePath:             "CGA-1234.json",
+				KnownID:                "CGA-1234",
+				KnownModified:          tt.knownModified,
+				Format:                 RecordFormatJSON,
+				Action:                 ActionImport,
+				CompareAgainstDatabase: true,
+			})
+
+			if opened != tt.wantOpened {
+				t.Errorf("SourceRecord.Open called = %v, want %v", opened, tt.wantOpened)
+			}
+			if (len(pub.Messages) == 1) != tt.wantPublished {
+				t.Errorf("published %d messages, wantPublished %v", len(pub.Messages), tt.wantPublished)
+			}
+		})
+	}
+}

go/internal/importer/mock_test.go

@@ -58,9 +58,13 @@ func (m *mockVulnerabilityStore) GetSourceModified(_ context.Context, vuln strin
 type mockSourceRecord struct {
 	DataToRead []byte
 	ReadError  error
+	OnOpen     func()
 }
 
 func (m mockSourceRecord) Open(_ context.Context) (io.ReadCloser, error) {
+	if m.OnOpen != nil {
+		m.OnOpen()
+	}
 	if m.ReadError != nil {
 		return nil, m.ReadError
 	}

go/internal/importer/rest.go

@@ -15,6 +15,11 @@ import (
 	"github.com/tidwall/gjson"
 )
 
+// restImportLookbackWindow overlaps incremental REST index scans with the previous
+// LastUpdated watermark so advisories published out of order or whose individual
+// record endpoint briefly lagged behind the index are still picked up.
+const restImportLookbackWindow = 12 * time.Hour
+
 type restSourceRecord struct {
 	cl      *http.Client
 	urlBase string
@@ -56,19 +61,25 @@ func handleImportREST(ctx context.Context, ch chan<- WorkItem, config Config, so
 	compiledIgnorePatterns := compileIgnorePatterns(sourceRepo)
 	hasUpdateTime := false
 	var lastUpdated time.Time
+	var cutoff time.Time
 	if !sourceRepo.REST.IgnoreLastImportTime && sourceRepo.REST.LastUpdated != nil {
 		lastUpdated = *sourceRepo.REST.LastUpdated
+		cutoff = lastUpdated.Add(-restImportLookbackWindow)
 		hasUpdateTime = true
 	}
 	timeOfRun := time.Now()
-	var lastModTime time.Time
+	var httpLastModified time.Time
 	if hasUpdateTime {
 		var err error
-		lastModTime, err = checkHEAD(ctx, config, sourceRepo)
+		httpLastModified, err = checkHEAD(ctx, config, sourceRepo)
 		if err != nil {
 			return err
 		}
-		if !lastModTime.IsZero() && (lastModTime.Before(lastUpdated) || lastModTime.Equal(lastUpdated)) {
+		prevHTTPLastModified := lastUpdated
+		if sourceRepo.REST.HTTPLastModified != nil {
+			prevHTTPLastModified = *sourceRepo.REST.HTTPLastModified
+		}
+		if !httpLastModified.IsZero() && !httpLastModified.After(prevHTTPLastModified) {
 			logger.InfoContext(ctx, "No changes since last update.",
 				slog.String("source", sourceRepo.Name),
 				slog.String("url", sourceRepo.REST.URL))
@@ -92,6 +103,9 @@ func handleImportREST(ctx context.Context, ch chan<- WorkItem, config Config, so
 
 		return fmt.Errorf("failed to fetch REST API: %d for %s", resp.StatusCode, sourceRepo.REST.URL)
 	}
+	if getHTTPLastModified, err := parseHTTPLastModified(resp.Header); err == nil && !getHTTPLastModified.IsZero() {
+		httpLastModified = getHTTPLastModified
+	}
 	data, err := io.ReadAll(resp.Body)
 	if err != nil {
 		logger.ErrorContext(ctx, "Failed to read REST API response", slog.String("source", sourceRepo.Name), slog.String("url", sourceRepo.REST.URL), slog.Any("error", err))
@@ -132,7 +146,7 @@ func handleImportREST(ctx context.Context, ch chan<- WorkItem, config Config, so
 		if mod.After(maxModified) {
 			maxModified = mod
 		}
-		if hasUpdateTime && (mod.Before(lastUpdated) || mod.Equal(lastUpdated)) {
+		if hasUpdateTime && !mod.After(cutoff) {
 			return true
 		}
 		if shouldIgnore(id.String(), sourceRepo.IDPrefixes, compiledIgnorePatterns) {
@@ -156,6 +170,8 @@ func handleImportREST(ctx context.Context, ch chan<- WorkItem, config Config, so
 			},
 			SourceRepository:       sourceRepo.Name,
 			SourcePath:             path,
+			KnownID:                id.String(),
+			KnownModified:          mod,
 			CompareAgainstDatabase: hasUpdateTime,
 			Format:                 RecordFormatJSON,
 			KeyPath:                sourceRepo.KeyPath,
@@ -181,10 +197,16 @@ func handleImportREST(ctx context.Context, ch chan<- WorkItem, config Config, so
 	if !maxModified.IsZero() && maxModified.Before(timeToUpdate) {
 		timeToUpdate = maxModified
 	}
-	if !lastModTime.IsZero() && lastModTime.Before(timeToUpdate) {
-		timeToUpdate = lastModTime
+	if !httpLastModified.IsZero() && httpLastModified.Before(timeToUpdate) {
+		timeToUpdate = httpLastModified
 	}
 	sourceRepo.REST.LastUpdated = &timeToUpdate
+	if !httpLastModified.IsZero() {
+		modUTC := httpLastModified.UTC()
+		sourceRepo.REST.HTTPLastModified = &modUTC
+	} else {
+		sourceRepo.REST.HTTPLastModified = nil
+	}
 	sourceRepo.REST.IgnoreLastImportTime = false
 	if err := config.SourceRepoStore.Update(ctx, sourceRepo.Name, sourceRepo); err != nil {
 		logger.ErrorContext(ctx, "Failed to update source repository", slog.Any("error", err), slog.String("source", sourceRepo.Name))
@@ -322,6 +344,19 @@ func handleDeleteREST(ctx context.Context, ch chan<- WorkItem, config Config, so
 	return nil
 }
 
+func parseHTTPLastModified(h http.Header) (time.Time, error) {
+	httpLastModified := h.Get("Last-Modified")
+	if httpLastModified == "" {
+		return time.Time{}, nil
+	}
+	t, err := http.ParseTime(httpLastModified)
+	if err != nil {
+		return time.Time{}, err
+	}
+
+	return t.UTC(), nil
+}
+
 // checkHEAD performs a HEAD request to check for updates
 func checkHEAD(ctx context.Context, config Config, sourceRepo *models.SourceRepository) (time.Time, error) {
 	req, err := http.NewRequest(http.MethodHead, sourceRepo.REST.URL, nil)
@@ -343,16 +378,15 @@ func checkHEAD(ctx context.Context, config Config, sourceRepo *models.SourceRepo
 		return time.Time{}, nil
 	}
 
-	lastModified := resp.Header.Get("Last-Modified")
-	if lastModified == "" {
+	if resp.Header.Get("Last-Modified") == "" {
 		logger.WarnContext(ctx, "No Last-Modified header found.",
 			slog.String("source", sourceRepo.Name),
 			slog.String("url", sourceRepo.REST.URL))
 
 		return time.Time{}, nil
 	}
 
-	lastModTime, err := time.Parse(time.RFC1123, lastModified)
+	httpLastModified, err := parseHTTPLastModified(resp.Header)
 	if err != nil {
 		logger.WarnContext(ctx, "Failed to parse Last-Modified header.",
 			slog.String("source", sourceRepo.Name),
@@ -362,7 +396,7 @@ func checkHEAD(ctx context.Context, config Config, sourceRepo *models.SourceRepo
 		return time.Time{}, nil
 	}
 
-	return lastModTime, nil
+	return httpLastModified, nil
 }
 
 func handleReconcileREST(ctx context.Context, ch chan<- WorkItem, config Config, sourceRepo *models.SourceRepository, timeOfRun time.Time) error {