fix(unified-storage): unlist the deleted resource's versions (#102588)

pkg/storage/unified/sql/backend.go

@@ -521,22 +521,19 @@ func (b *backend) ReadResource(ctx context.Context, req *resource.ReadRequest) *
 
 	// TODO: validate key ?
 
+	if req.ResourceVersion > 0 {
+		return b.readHistory(ctx, req.Key, req.ResourceVersion)
+	}
+
 	readReq := &sqlResourceReadRequest{
 		SQLTemplate: sqltemplate.New(b.dialect),
 		Request:     req,
 		Response:    NewReadResponse(),
 	}
-
-	sr := sqlResourceRead
-	if req.ResourceVersion > 0 {
-		// read a specific version
-		sr = sqlResourceHistoryRead
-	}
-
 	var res *resource.BackendReadResponse
 	err := b.db.WithTx(ctx, ReadCommittedRO, func(ctx context.Context, tx db.Tx) error {
 		var err error
-		res, err = dbutil.QueryRow(ctx, tx, sr, readReq)
+		res, err = dbutil.QueryRow(ctx, tx, sqlResourceRead, readReq)
 		return err
 	})
 
@@ -745,7 +742,38 @@ func (b *backend) listAtRevision(ctx context.Context, req *resource.ListRequest,
 	return iter.listRV, err
 }
 
-// getHistory fetches the resources from the resource table.
+// readHistory fetches the resource history from the resource_history table.
+func (b *backend) readHistory(ctx context.Context, key *resource.ResourceKey, rv int64) *resource.BackendReadResponse {
+	_, span := b.tracer.Start(ctx, tracePrefix+".ReadHistory")
+	defer span.End()
+
+	readReq := &sqlResourceHistoryReadRequest{
+		SQLTemplate: sqltemplate.New(b.dialect),
+		Request: &historyReadRequest{
+			Key:             key,
+			ResourceVersion: rv,
+		},
+		Response: NewReadResponse(),
+	}
+
+	var res *resource.BackendReadResponse
+	err := b.db.WithTx(ctx, ReadCommittedRO, func(ctx context.Context, tx db.Tx) error {
+		var err error
+		res, err = dbutil.QueryRow(ctx, tx, sqlResourceHistoryRead, readReq)
+		return err
+	})
+
+	if errors.Is(err, sql.ErrNoRows) {
+		return &resource.BackendReadResponse{Error: resource.NewNotFoundError(key)}
+	}
+	if err != nil {
+		return &resource.BackendReadResponse{Error: resource.AsErrorResult(err)}
+	}
+
+	return res
+}
+
+// getHistory fetches the resource history from the resource_history table.
 func (b *backend) getHistory(ctx context.Context, req *resource.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
 	ctx, span := b.tracer.Start(ctx, tracePrefix+"getHistory")
 	defer span.End()
@@ -784,6 +812,9 @@ func (b *backend) getHistory(ctx context.Context, req *resource.ListRequest, cb
 		listReq.MinRV = req.ResourceVersion
 	}
 
+	// Ignore last deleted history record when listing the trash, using exact matching or not older than matching with a specific RV
+	useLatestDeletionAsMinRV := listReq.MinRV == 0 && !listReq.Trash && req.VersionMatchV2 != resource.ResourceVersionMatchV2_Exact
+
 	err := b.db.WithTx(ctx, ReadCommittedRO, func(ctx context.Context, tx db.Tx) error {
 		var err error
 		iter.listRV, err = b.fetchLatestRV(ctx, tx, b.dialect, req.Options.Key.Group, req.Options.Key.Resource)
@@ -791,6 +822,14 @@ func (b *backend) getHistory(ctx context.Context, req *resource.ListRequest, cb
 			return err
 		}
 
+		if useLatestDeletionAsMinRV {
+			latestDeletedRV, err := b.fetchLatestHistoryRV(ctx, tx, b.dialect, req.Options.Key, resource.WatchEvent_DELETED)
+			if err != nil {
+				return err
+			}
+			listReq.MinRV = latestDeletedRV + 1
+		}
+
 		rows, err := dbutil.QueryRows(ctx, tx, sqlResourceHistoryGet, listReq)
 		if rows != nil {
 			defer func() {
@@ -860,3 +899,23 @@ func (b *backend) fetchLatestRV(ctx context.Context, x db.ContextExecer, d sqlte
 	}
 	return res.ResourceVersion, nil
 }
+
+// fetchLatestHistoryRV returns the current maximum RV in the resource_history table
+func (b *backend) fetchLatestHistoryRV(ctx context.Context, x db.ContextExecer, d sqltemplate.Dialect, key *resource.ResourceKey, eventType resource.WatchEvent_Type) (int64, error) {
+	ctx, span := b.tracer.Start(ctx, tracePrefix+"fetchLatestHistoryRV")
+	defer span.End()
+	res, err := dbutil.QueryRow(ctx, x, sqlResourceHistoryReadLatestRV, sqlResourceHistoryReadLatestRVRequest{
+		SQLTemplate: sqltemplate.New(d),
+		Request: &historyReadLatestRVRequest{
+			Key:       key,
+			EventType: eventType,
+		},
+		Response: new(resourceHistoryReadLatestRVResponse),
+	})
+	if errors.Is(err, sql.ErrNoRows) {
+		return 0, nil
+	} else if err != nil {
+		return 0, fmt.Errorf("get resource version: %w", err)
+	}
+	return res.ResourceVersion, nil
+}