go-transaction-manager

1
package manager
2

3
import (
4
	"context"
5
	"errors"
6
	"fmt"
7

8
	"go.uber.org/multierr"
9

10
	"github.com/avito-tech/go-transaction-manager/trm/v2"
11
)
12

13
// Closer closes trm.Transaction.
14
type Closer func(context.Context, interface{}, *error) error
15

16
type trCloser struct {
17
	tr     trm.Transaction
18
	cancel context.CancelFunc
19
	log    logger
20
}
21

22
func newTxCommit(tr trm.Transaction, l logger, c context.CancelFunc) Closer {
23
	return (&trCloser{
24
		tr:     tr,
25
		cancel: c,
26
		log:    l,
27
	}).close
28
}
29

30
//nolint:funlen
31
func (c *trCloser) close(ctx context.Context, p interface{}, errInProcessTr *error) error {
32
	defer c.cancel()
33

34
	// recovering from panic
35
	if p != nil {
36
		if c.tr.IsActive() {
37
			if err := c.tr.Rollback(ctx); err != nil {
38
				c.log.Warning(ctx, fmt.Sprintf("%v, %v", err, p))
39
			}
40
		}
41

42
		panic(p)
43
	}
44

45
	hasError := *errInProcessTr != nil
46
	isErrSkippable := hasError && trm.IsSkippable(*errInProcessTr)
47
	// TODO not sure that context errors should be propagated.
48
	isCtxCanceled := errors.Is(*errInProcessTr, context.Canceled)
49
	isCtxDeadlineExceeded := errors.Is(*errInProcessTr, context.DeadlineExceeded)
50
	isCtxErr := isCtxCanceled || isCtxDeadlineExceeded
51

52
	ctxErr := ctx.Err()
53

54
	if ctxErr != nil {
55
		if !hasError {
56
			*errInProcessTr = ctxErr
57
		} else if !isCtxCanceled && errors.Is(ctxErr, context.Canceled) ||
58
			!isCtxDeadlineExceeded && errors.Is(ctxErr, context.DeadlineExceeded) {
59
			*errInProcessTr = multierr.Combine(*errInProcessTr, ctxErr)
60
		}
61

62
		isCtxErr = true
63
		hasError = true
64
	}
65

66
	if !c.tr.IsActive() {
67
		if hasError {
68
			if isCtxErr || errors.Is(*errInProcessTr, trm.ErrAlreadyClosed) {
69
				return *errInProcessTr
70
			}
71

72
			return multierr.Combine(*errInProcessTr, trm.ErrAlreadyClosed)
73
		}
74

75
		return trm.ErrAlreadyClosed
76
	}
77

78
	if hasError && !isErrSkippable {
79
		if errRollback := c.tr.Rollback(ctx); errRollback != nil {
80
			return multierr.Combine(*errInProcessTr, trm.ErrRollback, errRollback)
81
		}
82

83
		return *errInProcessTr
84
	}
85

86
	if err := c.tr.Commit(ctx); err != nil {
87
		var errUnSkipped error
88
		if isErrSkippable {
89
			errUnSkipped = trm.UnSkippable(*errInProcessTr)
90
		}
91

92
		return multierr.Combine(trm.ErrCommit, err, errUnSkipped)
93
	} else if isErrSkippable {
94
		return *errInProcessTr
95
	}
96

97
	return nil
98
}
99

100
func newNilClose(cancel context.CancelFunc) Closer {
101
	return func(_ context.Context, p interface{}, err *error) error {
102
		defer cancel()
103

104
		if p != nil {
105
			panic(p)
106
		}
107

108
		if *err != nil {
109
			return *err
110
		}
111

112
		return nil
113
	}
114
}
115