<commit_before>// Copyright 2020 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 eventbox batches incoming events for a single Datastore entity
// for processing.
package eventbox

import (
	"context"
	"time"

	"github.com/google/uuid"
	"golang.org/x/sync/errgroup"

	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/logging"
	"go.chromium.org/luci/common/retry/transient"
	"go.chromium.org/luci/common/trace"
	"go.chromium.org/luci/gae/service/datastore"

	"go.chromium.org/luci/cv/internal/eventbox/dsset"
)

// Emit emits a new event with provided value and auto-generated unique ID.
func Emit(ctx context.Context, value []byte, to *datastore.Key) error {
	d := dsset.Set{Parent: to} // TombstonesDelay doesn't matter for Add.
	id := uuid.New().String()
	if err := d.Add(ctx, []dsset.Item{{ID: id, Value: value}}); err != nil {
		return errors.Annotate(err, "failed to send event").Err()
	}
	return nil
}

const tombstonesDelay = 5 * time.Minute

// List returns unprocessed events. For use in tests only.
func List(ctx context.Context, recipient *datastore.Key) (Events, error) {
	d := dsset.Set{
		Parent:          recipient,
		TombstonesDelay: tombstonesDelay,
	}
	switch l, err := d.List(ctx); {
	case err != nil:
		return nil, err
	default:
		return toEvents(l.Items), nil
	}
}

// ErrConcurretMutation indicates concurrent mutation.
var ErrConcurretMutation = errors.New("Concurrent mutation")

// IsErrConcurretMutation checks if error is due to concurrent mutation.
func IsErrConcurretMutation(err error) bool {
	ret := false
	errors.WalkLeaves(err, func(e error) bool {
		ret = (e == ErrConcurretMutation)
		return !ret // return false means stop traversing.
	})
	return ret
}

// ProcessBatch reliably processes events, while transactionally modifying state
// and performing arbitrary side effects.
//
// Returns
//  - a slice of non-nil post process functions which SHOULD be executed
//    immediately after calling this function. Those are generally extra work
//    that needs to be done as the result of state modification.
//  - error while processing events. Returns wrapped ErrConcurretMutation
//    if entity's EVersion has changed.
func ProcessBatch(ctx context.Context, recipient *datastore.Key, p Processor) ([]PostProcessFn, error) {
	ctx, span := trace.StartSpan(ctx, "go.chromium.org/luci/cv/internal/eventbox/ProcessBatch")
	var err error
	span.Attribute("recipient", recipient.String())
	defer func() { span.End(err) }()
	postProcessFn, err := processBatch(ctx, recipient, p)
	return postProcessFn, err
}

func processBatch(ctx context.Context, recipient *datastore.Key, p Processor) ([]PostProcessFn, error) {
	var state State
	var expectedEV EVersion
	eg, ectx := errgroup.WithContext(ctx)
	eg.Go(func() (err error) {
		state, expectedEV, err = p.LoadState(ectx)
		return
	})
	d := dsset.Set{
		Parent:          recipient,
		TombstonesDelay: tombstonesDelay,
	}
	var listing *dsset.Listing
	eg.Go(func() (err error) {
		if listing, err = d.List(ectx); err == nil {
			err = dsset.CleanupGarbage(ectx, listing.Garbage)
		}
		return
	})
	if err := eg.Wait(); err != nil {
		return nil, err
	}

	// Compute resulting state before transaction.
	transitions, garbage, err := p.Mutate(ctx, toEvents(listing.Items), state)
	if err != nil {
		return nil, err
	}
	if err := deleteSemanticGarbage(ctx, &d, garbage); err != nil {
		return nil, err
	}
	transitions = withoutNoops(transitions, state)
	if len(transitions) == 0 {
		return nil, nil // nothing to do.
	}

	var innerErr error
	var postProcessFns []PostProcessFn
	err = datastore.RunInTransaction(ctx, func(ctx context.Context) (err error) {
		defer func() { innerErr = err }()

		switch latestEV, err := p.FetchEVersion(ctx); {
		case err != nil:
			return err
		case latestEV != expectedEV:
			return errors.Annotate(ErrConcurretMutation, "EVersion read %d, but expected %d", latestEV, expectedEV).Tag(transient.Tag).Err()<commit_msg>cv: avoid appending to postProcessFns when transaction failure<commit_after>// Copyright 2020 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 eventbox batches incoming events for a single Datastore entity
// for processing.
package eventbox

import (
	"context"
	"time"

	"github.com/google/uuid"
	"golang.org/x/sync/errgroup"

	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/logging"
	"go.chromium.org/luci/common/retry/transient"
	"go.chromium.org/luci/common/trace"
	"go.chromium.org/luci/gae/service/datastore"

	"go.chromium.org/luci/cv/internal/eventbox/dsset"
)

// Emit emits a new event with provided value and auto-generated unique ID.
func Emit(ctx context.Context, value []byte, to *datastore.Key) error {
	d := dsset.Set{Parent: to} // TombstonesDelay doesn't matter for Add.
	id := uuid.New().String()
	if err := d.Add(ctx, []dsset.Item{{ID: id, Value: value}}); err != nil {
		return errors.Annotate(err, "failed to send event").Err()
	}
	return nil
}

const tombstonesDelay = 5 * time.Minute

// List returns unprocessed events. For use in tests only.
func List(ctx context.Context, recipient *datastore.Key) (Events, error) {
	d := dsset.Set{
		Parent:          recipient,
		TombstonesDelay: tombstonesDelay,
	}
	switch l, err := d.List(ctx); {
	case err != nil:
		return nil, err
	default:
		return toEvents(l.Items), nil
	}
}

// ErrConcurretMutation indicates concurrent mutation.
var ErrConcurretMutation = errors.New("Concurrent mutation")

// IsErrConcurretMutation checks if error is due to concurrent mutation.
func IsErrConcurretMutation(err error) bool {
	ret := false
	errors.WalkLeaves(err, func(e error) bool {
		ret = (e == ErrConcurretMutation)
		return !ret // return false means stop traversing.
	})
	return ret
}

// ProcessBatch reliably processes events, while transactionally modifying state
// and performing arbitrary side effects.
//
// Returns
//  - a slice of non-nil post process functions which SHOULD be executed
//    immediately after calling this function. Those are generally extra work
//    that needs to be done as the result of state modification.
//  - error while processing events. Returns wrapped ErrConcurretMutation
//    if entity's EVersion has changed.
func ProcessBatch(ctx context.Context, recipient *datastore.Key, p Processor) ([]PostProcessFn, error) {
	ctx, span := trace.StartSpan(ctx, "go.chromium.org/luci/cv/internal/eventbox/ProcessBatch")
	var err error
	span.Attribute("recipient", recipient.String())
	defer func() { span.End(err) }()
	postProcessFn, err := processBatch(ctx, recipient, p)
	return postProcessFn, err
}

func processBatch(ctx context.Context, recipient *datastore.Key, p Processor) ([]PostProcessFn, error) {
	var state State
	var expectedEV EVersion
	eg, ectx := errgroup.WithContext(ctx)
	eg.Go(func() (err error) {
		state, expectedEV, err = p.LoadState(ectx)
		return
	})
	d := dsset.Set{
		Parent:          recipient,
		TombstonesDelay: tombstonesDelay,
	}
	var listing *dsset.Listing
	eg.Go(func() (err error) {
		if listing, err = d.List(ectx); err == nil {
			err = dsset.CleanupGarbage(ectx, listing.Garbage)
		}
		return
	})
	if err := eg.Wait(); err != nil {
		return nil, err
	}

	// Compute resulting state before transaction.
	transitions, garbage, err := p.Mutate(ctx, toEvents(listing.Items), state)
	if err != nil {
		return nil, err
	}
	if err := deleteSemanticGarbage(ctx, &d, garbage); err != nil {
		return nil, err
	}
	transitions = withoutNoops(transitions, state)
	if len(transitions) == 0 {
		return nil, nil // nothing to do.
	}

	var innerErr error
	var postProcessFns []PostProcessFn
	err = datastore.RunInTransaction(ctx, func(ctx context.Context) (err error) {
		defer func() { innerErr = err }()
		postProcessFns = nil //  reset, since this func can be retried

		switch latestEV, err := p.FetchEVersion(ctx); {
		case err != nil:
			return err
		case latestEV != expectedEV:
			return errors.Annotate(ErrConcurretMutation, "EVersion read %d, but expected %d", latestEV, expectedEV).Tag(transient.Tag).Err()
{'fraction_non_alphanumeric': 0.11697905181918412, 'fraction_numerical': 0.002205071664829107, 'mean_word_length': 3.946019629225736, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '635390', 'n_tokens_mistral': 3011, 'n_tokens_neox': 2879, 'n_words': 925}