Use created_at field to determine if entry is stale

Signed-off-by: Sorin Dumitru <[email protected]>

pkg/agent/client/client.go

@@ -39,6 +39,7 @@ var (
 		RevisionNumber: true,
 		StoreSvid:      true,
 		Hint:           true,
+		CreatedAt:      true,
 	}
 )
 
@@ -416,6 +417,23 @@ func (c *client) syncEntries(ctx context.Context, cachedEntries map[string]*comm
 	return stats, nil
 }
 
+func entryIsStale(entry *common.RegistrationEntry, revisionNumber, revisionCreatedAt int64) bool {
+	if entry.RevisionNumber != revisionNumber {
+		return true
+	}
+
+	// TOOD: remove in SPIRE 1.14
+	if revisionCreatedAt == 0 {
+		return false
+	}
+
+	if entry.CreatedAt != revisionCreatedAt {
+		return true
+	}
+
+	return false
+}
+
 func (c *client) streamAndSyncEntries(ctx context.Context, entryClient entryv1.EntryClient, cachedEntries map[string]*common.RegistrationEntry) (stats SyncEntriesStats, err error) {
 	// Build a set of all the entries to be removed. This set is initialized
 	// with all entries currently known. As entries are synced down from the
@@ -459,7 +477,7 @@ func (c *client) streamAndSyncEntries(ctx context.Context, entryClient entryv1.E
 			// If entry is either not cached or is stale, record the ID so
 			// the full entry can be requested after syncing down all
 			// entry revisions.
-			if cachedEntry, ok := cachedEntries[entryRevision.Id]; !ok || cachedEntry.RevisionNumber < entryRevision.RevisionNumber {
+			if cachedEntry, ok := cachedEntries[entryRevision.Id]; !ok || entryIsStale(cachedEntry, entryRevision.GetRevisionNumber(), entryRevision.GetCreatedAt()) {
 				needFull = append(needFull, entryRevision.Id)
 			}
 		}
@@ -487,7 +505,7 @@ func (c *client) streamAndSyncEntries(ctx context.Context, entryClient entryv1.E
 			switch {
 			case !ok:
 				stats.Missing++
-			case cachedEntry.RevisionNumber < entry.RevisionNumber:
+			case entryIsStale(cachedEntry, entry.GetRevisionNumber(), entry.GetCreatedAt()):
 				stats.Stale++
 			}
 

pkg/agent/client/client_test.go

@@ -285,14 +285,19 @@ func TestSyncUpdatesEntries(t *testing.T) {
 		assert.Equal(t, expected, cachedEntries)
 	}
 
-	entryA1 := makeEntry("A", 1)
-	entryB1 := makeEntry("B", 1)
-	entryC1 := makeEntry("C", 1)
-	entryD1 := makeEntry("D", 1)
+	firstDate := time.Date(2024, time.December, 31, 0, 0, 0, 0, time.UTC)
+	secondDate := time.Date(2025, time.January, 1, 0, 0, 0, 0, time.UTC)
 
-	entryA2 := makeEntry("A", 2)
-	entryB2 := makeEntry("B", 2)
-	entryC2 := makeEntry("C", 2)
+	entryA1 := makeEntry("A", 1, firstDate)
+	entryB1 := makeEntry("B", 1, firstDate)
+	entryC1 := makeEntry("C", 1, firstDate)
+	entryD1 := makeEntry("D", 1, firstDate)
+
+	entryA2 := makeEntry("A", 2, firstDate)
+	entryB2 := makeEntry("B", 2, firstDate)
+	entryC2 := makeEntry("C", 2, firstDate)
+
+	entryB1prime := makeEntry("B", 1, secondDate)
 
 	// No entries yet
 	syncAndAssertEntries(t, 0, 0, 0, 0)
@@ -314,6 +319,12 @@ func TestSyncUpdatesEntries(t *testing.T) {
 
 	// Sync again but with no changes.
 	syncAndAssertEntries(t, 3, 0, 0, 0, entryA2, entryB2, entryC2)
+
+	// Sync again after recreating an entry with the same entry ID, which should be marked stale
+	syncAndAssertEntries(t, 3, 0, 1, 0, entryA2, entryB1prime, entryC2)
+
+	// Sync again after the database has been rolled back to a previous version
+	syncAndAssertEntries(t, 4, 1, 3, 0, entryA1, entryB1, entryC1, entryD1)
 }
 
 func TestRenewSVID(t *testing.T) {
@@ -1185,16 +1196,7 @@ type testServer struct {
 }
 
 func checkAuthorizedEntryOutputMask(outputMask *types.EntryMask) error {
-	if diff := cmp.Diff(outputMask, &types.EntryMask{
-		SpiffeId:       true,
-		Selectors:      true,
-		FederatesWith:  true,
-		Admin:          true,
-		Downstream:     true,
-		RevisionNumber: true,
-		StoreSvid:      true,
-		Hint:           true,
-	}, protocmp.Transform()); diff != "" {
+	if diff := cmp.Diff(outputMask, entryOutputMask, protocmp.Transform()); diff != "" {
 		return status.Errorf(codes.InvalidArgument, "invalid output mask requested: %s", diff)
 	}
 	return nil
@@ -1214,12 +1216,13 @@ func makeCommonBundle(trustDomainName string) *common.Bundle {
 	}
 }
 
-func makeEntry(id string, revisionNumber int64) *types.Entry {
+func makeEntry(id string, revisionNumber int64, createdAt time.Time) *types.Entry {
 	return &types.Entry{
 		Id:             id,
 		SpiffeId:       &types.SPIFFEID{TrustDomain: "example.org", Path: "/workload"},
 		ParentId:       &types.SPIFFEID{TrustDomain: "example.org", Path: "/agent"},
 		Selectors:      []*types.Selector{{Type: "not", Value: "relevant"}},
 		RevisionNumber: revisionNumber,
+		CreatedAt:      createdAt.Unix(),
 	}
 }

pkg/agent/client/util.go

@@ -81,5 +81,6 @@ func slicedEntryFromProto(e *types.Entry) (*common.RegistrationEntry, error) {
 		Admin:          e.Admin,
 		Downstream:     e.Downstream,
 		Hint:           e.Hint,
+		CreatedAt:      e.CreatedAt,
 	}, nil
 }

pkg/server/api/entry/v1/service.go

@@ -476,6 +476,7 @@ func SyncAuthorizedEntries(stream entryv1.Entry_SyncAuthorizedEntriesServer, ent
 		for j, entry := range entries[i : i+n] {
 			entryRevisions[j].Id = entry.Id
 			entryRevisions[j].RevisionNumber = entry.RevisionNumber
+			entryRevisions[j].CreatedAt = entry.CreatedAt
 		}
 
 		if err := stream.Send(&entryv1.SyncAuthorizedEntriesResponse{

pkg/server/api/entry/v1/service_test.go

@@ -3142,6 +3142,7 @@ func TestSyncAuthorizedEntries(t *testing.T) {
 								ParentId:       entry1.ParentId,
 								Selectors:      entry1.Selectors,
 								RevisionNumber: entry1.RevisionNumber,
+								CreatedAt:      entry1.CreatedAt,
 							},
 							{
 								Id:             entry2.Id,