fix race condition in watcher due to pruning

Signed-off-by: Satyam Bhardwaj <sabhardw@redhat.com>

rh-pre-commit.version: 2.1.0
rh-pre-commit.check-secrets: ENABLED

pkg/watcher/reconciler/dynamic/dynamic.go

@@ -229,6 +229,23 @@ func (r *Reconciler) deleteUponCompletion(ctx context.Context, o results.Object)
 		return controller.NewRequeueAfter(r.cfg.RequeueInterval)
 	}
 
+	if r.resultsClient.LogsClient != nil {
+		logStored, err := r.resultsClient.LogStatus(ctx, o)
+		if err != nil {
+			logger.Errorw("Error confirming object's logs were stored - requeuing to prune later",
+				zap.String("namespace", o.GetNamespace()),
+				zap.String("kind", o.GetObjectKind().GroupVersionKind().Kind),
+				zap.String("name", o.GetName()),
+				zap.Error(err),
+			)
+			return controller.NewRequeueAfter(r.cfg.RequeueInterval)
+		}
+		if !logStored {
+			logger.Debugw("Object is streaming logs - requeuing to prune later", zap.Duration("results.tekton.dev/requeueAfter", r.cfg.RequeueInterval))
+			return controller.NewRequeueAfter(r.cfg.RequeueInterval)
+		}
+	}
+
 	logger.Infow("Deleting object", zap.String("results.tekton.dev/uid", string(o.GetUID())),
 		zap.Int64("results.tekton.dev/time-taken-seconds", int64(time.Since(*completionTime).Seconds())))
 
@@ -341,6 +358,7 @@ func (r *Reconciler) sendLog(ctx context.Context, o results.Object) error {
 					zap.String("name", o.GetName()),
 					zap.Error(err),
 				)
+				return
 			}
 			logger.Debugw("Streaming log completed",
 				zap.String("namespace", o.GetNamespace()),

pkg/watcher/reconciler/dynamic/dynamic_test.go

@@ -207,6 +207,16 @@ func TestReconcile_TaskRun(t *testing.T) {
 	})
 
 	t.Run("delete object once grace period elapses", func(t *testing.T) {
+		// Recreate the object to retest the deletion
+		if err := trclient.Delete(ctx, taskrun.GetName(), metav1.DeleteOptions{}); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := trclient.Create(ctx, taskrun, metav1.CreateOptions{}); err != nil {
+			t.Fatal(err)
+		}
+		// disable logs client so that requeuing due to missing logs
+		// won't interfere with this test
+		r.resultsClient.LogsClient = nil
 		// Enable object deletion, re-reconcile
 		cfg.CompletedResourceGracePeriod = 1 * time.Second
 

pkg/watcher/results/logs.go

@@ -2,6 +2,8 @@ package results
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 
 	"github.com/google/uuid"
 	"github.com/tektoncd/results/pkg/api/server/v1alpha2/log"
@@ -86,3 +88,29 @@ func (c *Client) GetLogRecord(ctx context.Context, o Object) (*pb.Record, error)
 	}
 	return rec, err
 }
+
+// LogStatus checks if logs related to the given object have been successfully stored.
+func (c *Client) LogStatus(ctx context.Context, o Object) (bool, error) {
+	rec, err := c.GetLogRecord(ctx, o)
+	if err != nil {
+		return false, err
+	}
+
+	var logStatus LogStatus
+
+	err = json.Unmarshal(rec.GetData().GetValue(), &logStatus)
+	if err != nil {
+		return false, fmt.Errorf("error unmarshalling : %w", err)
+	}
+
+	return logStatus.Status.IsStored, nil
+}
+
+// LogStatus is a struct to match the JSON structure of the log status.
+type LogStatus struct {
+	Status `json:"status"`
+}
+
+type Status struct {
+	IsStored bool `json:"isStored"`
+}