-
Notifications
You must be signed in to change notification settings - Fork 4.2k
/
transactions.go
126 lines (115 loc) · 3.25 KB
/
transactions.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
package physical
import (
"context"
multierror "github.com/hashicorp/go-multierror"
)
// TxnEntry is an operation that takes atomically as part of
// a transactional update. Only supported by Transactional backends.
type TxnEntry struct {
Operation Operation
Entry *Entry
}
// Transactional is an optional interface for backends that
// support doing transactional updates of multiple keys. This is
// required for some features such as replication.
type Transactional interface {
// The function to run a transaction
Transaction(context.Context, []*TxnEntry) error
}
type PseudoTransactional interface {
// An internal function should do no locking or permit pool acquisition.
// Depending on the backend and if it natively supports transactions, these
// may simply chain to the normal backend functions.
GetInternal(context.Context, string) (*Entry, error)
PutInternal(context.Context, *Entry) error
DeleteInternal(context.Context, string) error
}
// Implements the transaction interface
func GenericTransactionHandler(ctx context.Context, t PseudoTransactional, txns []*TxnEntry) (retErr error) {
rollbackStack := make([]*TxnEntry, 0, len(txns))
var dirty bool
// We walk the transactions in order; each successful operation goes into a
// LIFO for rollback if we hit an error along the way
TxnWalk:
for _, txn := range txns {
switch txn.Operation {
case DeleteOperation:
entry, err := t.GetInternal(ctx, txn.Entry.Key)
if err != nil {
retErr = multierror.Append(retErr, err)
dirty = true
break TxnWalk
}
if entry == nil {
// Nothing to delete or roll back
continue
}
rollbackEntry := &TxnEntry{
Operation: PutOperation,
Entry: &Entry{
Key: entry.Key,
Value: entry.Value,
},
}
err = t.DeleteInternal(ctx, txn.Entry.Key)
if err != nil {
retErr = multierror.Append(retErr, err)
dirty = true
break TxnWalk
}
rollbackStack = append([]*TxnEntry{rollbackEntry}, rollbackStack...)
case PutOperation:
entry, err := t.GetInternal(ctx, txn.Entry.Key)
if err != nil {
retErr = multierror.Append(retErr, err)
dirty = true
break TxnWalk
}
// Nothing existed so in fact rolling back requires a delete
var rollbackEntry *TxnEntry
if entry == nil {
rollbackEntry = &TxnEntry{
Operation: DeleteOperation,
Entry: &Entry{
Key: txn.Entry.Key,
},
}
} else {
rollbackEntry = &TxnEntry{
Operation: PutOperation,
Entry: &Entry{
Key: entry.Key,
Value: entry.Value,
},
}
}
err = t.PutInternal(ctx, txn.Entry)
if err != nil {
retErr = multierror.Append(retErr, err)
dirty = true
break TxnWalk
}
rollbackStack = append([]*TxnEntry{rollbackEntry}, rollbackStack...)
}
}
// Need to roll back because we hit an error along the way
if dirty {
// While traversing this, if we get an error, we continue anyways in
// best-effort fashion
for _, txn := range rollbackStack {
switch txn.Operation {
case DeleteOperation:
err := t.DeleteInternal(ctx, txn.Entry.Key)
if err != nil {
retErr = multierror.Append(retErr, err)
}
case PutOperation:
err := t.PutInternal(ctx, txn.Entry)
if err != nil {
retErr = multierror.Append(retErr, err)
}
}
}
}
return
}