<commit_before>// Copyright 2021 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package tjupdate contains Updater, which handles an UpdateTryjobTask.
//
// This involves updating a Tryjob by querying the backend system, and updating
// Tryjob entities in Datastore.
package tjupdate

import (
	"context"
	"fmt"
	"strings"
	"sync"

	"google.golang.org/protobuf/proto"

	"go.chromium.org/luci/common/clock"
	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/retry/transient"
	"go.chromium.org/luci/gae/filter/txndefer"
	"go.chromium.org/luci/gae/service/datastore"

	"go.chromium.org/luci/cv/internal/common"
	"go.chromium.org/luci/cv/internal/metrics"
	"go.chromium.org/luci/cv/internal/rpc/versioning"
	"go.chromium.org/luci/cv/internal/run"
	"go.chromium.org/luci/cv/internal/tryjob"
)

type updaterBackend interface {
	// Kind identifies the backend.
	//
	// It's also the first part of the Tryjob's ExternalID, e.g. "buildbucket".
	// Must NOT contain a slash.
	Kind() string
	// Update should fetch the Tryjob given the current entity in Datastore.
	//
	// MUST NOT modify the given Tryjob object.
	Update(ctx context.Context, saved *tryjob.Tryjob) (tryjob.Status, *tryjob.Result, error)
}

// rmNotifier abstracts out the Run Manager's Notifier.
type rmNotifier interface {
	NotifyTryjobsUpdated(context.Context, common.RunID, *tryjob.TryjobUpdatedEvents) error
}

// Updater knows how to update Tryjobs, notifying other CV parts as needed.
type Updater struct {
	env        *common.Env
	rmNotifier rmNotifier

	rwmutex  sync.RWMutex // guards `backends`
	backends map[string]updaterBackend
}

// NewUpdater creates a new Updater.
//
// Starts without backends, but they should be added via RegisterBackend().
func NewUpdater(env *common.Env, tn *tryjob.Notifier, rm rmNotifier) *Updater {
	u := &Updater{
		env:        env,
		rmNotifier: rm,
		backends:   make(map[string]updaterBackend, 1),
	}
	tn.Bindings.Update.AttachHandler(func(ctx context.Context, payload proto.Message) error {
		return common.TQifyError(ctx, u.handleTask(ctx, payload.(*tryjob.UpdateTryjobTask)))
	})
	return u
}

// RegisterBackend registers a backend.
//
// Panics if backend for the same kind is already registered.
func (u *Updater) RegisterBackend(b updaterBackend) {
	kind := b.Kind()
	if strings.ContainsRune(kind, '/') {
		panic(fmt.Errorf("backend %T of kind %q must not contain '/'", b, kind))
	}
	u.rwmutex.Lock()
	defer u.rwmutex.Unlock()
	if _, exists := u.backends[kind]; exists {
		panic(fmt.Errorf("backend %q is already registered", kind))
	}
	u.backends[kind] = b
}

// handleTask handles an UpdateTryjobTask.
//
// This task involves checking the status of a Tryjob and updating its
// Datastore entity, and notifying Runs which care about this Tryjob.
func (u *Updater) handleTask(ctx context.Context, task *tryjob.UpdateTryjobTask) error {
	tj := &tryjob.Tryjob{ID: common.TryjobID(task.Id)}
	switch {
	case task.GetId() != 0:
		switch err := datastore.Get(ctx, tj); {
		case err == nil:
			if task.GetExternalId() != "" && task.GetExternalId() != string(tj.ExternalID) {
				return errors.Reason("the given internal and external IDs for the Tryjob do not match").Err()
			}
		case err == datastore.ErrNoSuchEntity:
			return errors.Annotate(err, "unknown Tryjob with ID %d", task.Id).Err()
		default:
			return errors.Annotate(err, "loading Tryjob with ID %d", task.Id).Tag(transient.Tag).Err()
		}
	case task.GetExternalId() != "":
		var err error
		switch tj, err = tryjob.ExternalID(task.ExternalId).Load(ctx); {
		case err != nil:
			return errors.Annotate(err, "loading Tryjob with ExternalID %s", task.ExternalId).Tag(transient.Tag).Err()
		case tj == nil:
			return errors.Reason("unknown Tryjob with ExternalID %s", task.ExternalId).Err()
		}
	default:
		return errors.Reason("expected at least one of {Id, ExternalId} in %+v", task).Err()
	}
	loadedEVer := tj.EVersion

	backend, err := u.backendFor(tj)
	if err != nil {
		return errors.Annotate(err, "resolving backend for %v", tj).Err()
	}

	status, result, err := backend.Update(ctx, tj)
	switch {
	case err != nil:
		return errors.Annotate(err, "reading status and result from %q", tj.ExternalID).Err()
	case status == tj.Status && proto.Equal(tj.Result, result):
		return nil
	}
	// Capture the error that may cause the transaction to commit, and any
	// relevant tags.
	var innerErr error
	err = datastore.RunInTransaction(ctx, func(ctx context.Context) (err error) {
		defer func() {
			innerErr = err
		}()
		tj := &tryjob.Tryjob{ID: common.TryjobID(tj.ID)}
		if err := datastore.Get(ctx, tj); err != nil {
			return errors.Annotate(err, "failed to load Tryjob %d", tj.ID).Tag(transient.Tag).Err()
		}
		if loadedEVer != tj.EVersion {
			// A parallel task must have already updated this Tryjob; retry.
			return errors.Reason("the tryjob data has changed").Tag(transient.Tag).Err()
		}

		if tj.LaunchedBy != "" && status == tryjob.Status_ENDED && tj.Status != status {
			// Tryjob launched by CV has transitioned to end status
			r := &run.Run{ID: tj.LaunchedBy}
			if err := datastore.Get(ctx, r); err != nil {
				return errors.Annotate(err, "failed to load Run %s", r.ID).Tag(transient.Tag).Err()
			}
			project, configGroup := r.ID.LUCIProject(), r.ConfigGroupID.Name()
			isRetry := true
			for _, exec := range r.Tryjobs.GetState().GetExecutions() {
				if len(exec.GetAttempts()) > 0 && exec.GetAttempts()[0].TryjobId == int64(tj.ID) {
					isRetry = false
					break
				}
			}
			txndefer.Defer(ctx, func(ctx context.Context) {
				tryjob.RunWithBuilderMetricsTarget(ctx, u.env, tj.Definition, func(ctx context.Context) {
					metrics.Public.TryjobEnded.Add(ctx, 1,
						project,<commit_msg>cv: make stale tryjob data a known retryable error<commit_after>// Copyright 2021 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package tjupdate contains Updater, which handles an UpdateTryjobTask.
//
// This involves updating a Tryjob by querying the backend system, and updating
// Tryjob entities in Datastore.
package tjupdate

import (
	"context"
	"fmt"
	"strings"
	"sync"

	"google.golang.org/protobuf/proto"

	"go.chromium.org/luci/common/clock"
	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/retry/transient"
	"go.chromium.org/luci/gae/filter/txndefer"
	"go.chromium.org/luci/gae/service/datastore"

	"go.chromium.org/luci/cv/internal/common"
	"go.chromium.org/luci/cv/internal/metrics"
	"go.chromium.org/luci/cv/internal/rpc/versioning"
	"go.chromium.org/luci/cv/internal/run"
	"go.chromium.org/luci/cv/internal/tryjob"
)

type updaterBackend interface {
	// Kind identifies the backend.
	//
	// It's also the first part of the Tryjob's ExternalID, e.g. "buildbucket".
	// Must NOT contain a slash.
	Kind() string
	// Update should fetch the Tryjob given the current entity in Datastore.
	//
	// MUST NOT modify the given Tryjob object.
	Update(ctx context.Context, saved *tryjob.Tryjob) (tryjob.Status, *tryjob.Result, error)
}

// rmNotifier abstracts out the Run Manager's Notifier.
type rmNotifier interface {
	NotifyTryjobsUpdated(context.Context, common.RunID, *tryjob.TryjobUpdatedEvents) error
}

// Updater knows how to update Tryjobs, notifying other CV parts as needed.
type Updater struct {
	env        *common.Env
	rmNotifier rmNotifier

	rwmutex  sync.RWMutex // guards `backends`
	backends map[string]updaterBackend
}

// NewUpdater creates a new Updater.
//
// Starts without backends, but they should be added via RegisterBackend().
func NewUpdater(env *common.Env, tn *tryjob.Notifier, rm rmNotifier) *Updater {
	u := &Updater{
		env:        env,
		rmNotifier: rm,
		backends:   make(map[string]updaterBackend, 1),
	}
	tn.Bindings.Update.AttachHandler(func(ctx context.Context, payload proto.Message) error {
		err := u.handleTask(ctx, payload.(*tryjob.UpdateTryjobTask))
		return common.TQIfy{
			KnownRetry: []error{errStaleTryjobData},
		}.Error(ctx, err)
	})
	return u
}

// RegisterBackend registers a backend.
//
// Panics if backend for the same kind is already registered.
func (u *Updater) RegisterBackend(b updaterBackend) {
	kind := b.Kind()
	if strings.ContainsRune(kind, '/') {
		panic(fmt.Errorf("backend %T of kind %q must not contain '/'", b, kind))
	}
	u.rwmutex.Lock()
	defer u.rwmutex.Unlock()
	if _, exists := u.backends[kind]; exists {
		panic(fmt.Errorf("backend %q is already registered", kind))
	}
	u.backends[kind] = b
}

// errStaleTryjobData may be returned by handle task if a concurrent task
// updates the Tryjob entity while this task is in the middle of update.
var errStaleTryjobData = errors.New("loaded stale Tryjob data", transient.Tag)

// handleTask handles an UpdateTryjobTask.
//
// This task involves checking the status of a Tryjob and updating its
// Datastore entity, and notifying Runs which care about this Tryjob.
func (u *Updater) handleTask(ctx context.Context, task *tryjob.UpdateTryjobTask) error {
	tj := &tryjob.Tryjob{ID: common.TryjobID(task.Id)}
	switch {
	case task.GetId() != 0:
		switch err := datastore.Get(ctx, tj); err {
		case nil:
			if task.GetExternalId() != "" && task.GetExternalId() != string(tj.ExternalID) {
				return errors.Reason("the given internal and external IDs for the Tryjob do not match").Err()
			}
		case datastore.ErrNoSuchEntity:
			return errors.Annotate(err, "unknown Tryjob with ID %d", task.Id).Err()
		default:
			return errors.Annotate(err, "loading Tryjob with ID %d", task.Id).Tag(transient.Tag).Err()
		}
	case task.GetExternalId() != "":
		var err error
		switch tj, err = tryjob.ExternalID(task.ExternalId).Load(ctx); {
		case err != nil:
			return errors.Annotate(err, "loading Tryjob with ExternalID %s", task.ExternalId).Tag(transient.Tag).Err()
		case tj == nil:
			return errors.Reason("unknown Tryjob with ExternalID %s", task.ExternalId).Err()
		}
	default:
		return errors.Reason("expected at least one of {Id, ExternalId} in %+v", task).Err()
	}
	loadedEVer := tj.EVersion

	backend, err := u.backendFor(tj)
	if err != nil {
		return errors.Annotate(err, "resolving backend for %v", tj).Err()
	}

	status, result, err := backend.Update(ctx, tj)
	switch {
	case err != nil:
		return errors.Annotate(err, "reading status and result from %q", tj.ExternalID).Err()
	case status == tj.Status && proto.Equal(tj.Result, result):
		return nil
	}
	// Capture the error that may cause the transaction to commit, and any
	// relevant tags.
	var innerErr error
	err = datastore.RunInTransaction(ctx, func(ctx context.Context) (err error) {
		defer func() {
			innerErr = err
		}()
		tj := &tryjob.Tryjob{ID: common.TryjobID(tj.ID)}
		if err := datastore.Get(ctx, tj); err != nil {
			return errors.Annotate(err, "failed to load Tryjob %d", tj.ID).Tag(transient.Tag).Err()
		}
		if loadedEVer != tj.EVersion {
			// A parallel task must have already updated this Tryjob; retry.
			return errStaleTryjobData
		}

		if tj.LaunchedBy != "" && status == tryjob.Status_ENDED && tj.Status != status {
			// Tryjob launched by CV has transitioned to end status
			r := &run.Run{ID: tj.LaunchedBy}
			if err := datastore.Get(ctx, r); err != nil {
				return errors.Annotate(err, "failed to load Run %s", r.ID).Tag(transient.Tag).Err()
			}
			project, configGroup := r.ID.LUCIProject(), r.ConfigGroupID.Name()
			isRetry := true
			for _, exec := range r.Tryjobs.GetState().GetExecutions() {
				if len(exec.GetAttempts()) > 0 && exec.GetAttempts()[0].TryjobId == int64(tj.ID) {
					isRetry = false
					break
				}
			}
			txndefer.Defer(ctx, func(ctx context.Context) {
				tryjob.RunWithBuilderMetricsTarget(ctx, u.env, tj.Definition, func(ctx context.Context) {
					metrics.Public.TryjobEnded.Add(ctx, 1,
						project,
{'fraction_non_alphanumeric': 0.12916666666666668, 'fraction_numerical': 0.0023584905660377358, 'mean_word_length': 3.833206686930091, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7571724', 'n_tokens_mistral': 4519, 'n_tokens_neox': 4264, 'n_words': 1232}