forked from distribution/distribution
/
transaction.go
241 lines (212 loc) · 7.32 KB
/
transaction.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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
// Copyright 2014 Google Inc. All Rights Reserved.
//
// 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 datastore
import (
"errors"
"net/http"
"github.com/golang/protobuf/proto"
"golang.org/x/net/context"
pb "google.golang.org/cloud/internal/datastore"
"google.golang.org/cloud/internal/transport"
)
// ErrConcurrentTransaction is returned when a transaction is rolled back due
// to a conflict with a concurrent transaction.
var ErrConcurrentTransaction = errors.New("datastore: concurrent transaction")
var errExpiredTransaction = errors.New("datastore: transaction expired")
// A TransactionOption configures the Transaction returned by NewTransaction.
type TransactionOption interface {
apply(*pb.BeginTransactionRequest)
}
type isolation struct {
level pb.BeginTransactionRequest_IsolationLevel
}
func (i isolation) apply(req *pb.BeginTransactionRequest) {
req.IsolationLevel = i.level.Enum()
}
var (
// Snapshot causes the transaction to enforce a snapshot isolation level.
Snapshot TransactionOption = isolation{pb.BeginTransactionRequest_SNAPSHOT}
// Serializable causes the transaction to enforce a serializable isolation level.
Serializable TransactionOption = isolation{pb.BeginTransactionRequest_SERIALIZABLE}
)
// Transaction represents a set of datastore operations to be committed atomically.
//
// Operations are enqueued by calling the Put and Delete methods on Transaction
// (or their Multi-equivalents). These operations are only committed when the
// Commit method is invoked. To ensure consistency, reads must be performed by
// using Transaction's Get method or by using the Transaction method when
// building a query.
//
// A Transaction must be committed or rolled back exactly once.
type Transaction struct {
id []byte
client *Client
ctx context.Context
mutation *pb.Mutation // The mutations to apply.
pending []*PendingKey // Incomplete keys pending transaction completion.
}
// NewTransaction starts a new transaction.
func (c *Client) NewTransaction(ctx context.Context, opts ...TransactionOption) (*Transaction, error) {
req, resp := &pb.BeginTransactionRequest{}, &pb.BeginTransactionResponse{}
for _, o := range opts {
o.apply(req)
}
if err := c.call(ctx, "beginTransaction", req, resp); err != nil {
return nil, err
}
return &Transaction{
id: resp.Transaction,
ctx: ctx,
client: c,
mutation: &pb.Mutation{},
}, nil
}
// Commit applies the enqueued operations atomically.
func (t *Transaction) Commit() (*Commit, error) {
if t.id == nil {
return nil, errExpiredTransaction
}
req := &pb.CommitRequest{
Transaction: t.id,
Mutation: t.mutation,
Mode: pb.CommitRequest_TRANSACTIONAL.Enum(),
}
t.id = nil
resp := &pb.CommitResponse{}
if err := t.client.call(t.ctx, "commit", req, resp); err != nil {
if e, ok := err.(*transport.ErrHTTP); ok && e.StatusCode == http.StatusConflict {
// TODO(jbd): Make sure that we explicitly handle the case where response
// has an HTTP 409 and the error message indicates that it's an concurrent
// transaction error.
return nil, ErrConcurrentTransaction
}
return nil, err
}
// Copy any newly minted keys into the returned keys.
if len(t.pending) != len(resp.MutationResult.InsertAutoIdKey) {
return nil, errors.New("datastore: internal error: server returned the wrong number of keys")
}
commit := &Commit{}
for i, p := range t.pending {
p.key = protoToKey(resp.MutationResult.InsertAutoIdKey[i])
p.commit = commit
}
return commit, nil
}
// Rollback abandons a pending transaction.
func (t *Transaction) Rollback() error {
if t.id == nil {
return errExpiredTransaction
}
id := t.id
t.id = nil
return t.client.call(t.ctx, "rollback", &pb.RollbackRequest{Transaction: id}, &pb.RollbackResponse{})
}
// Get is the transaction-specific version of the package function Get.
// All reads performed during the transaction will come from a single consistent
// snapshot. Furthermore, if the transaction is set to a serializable isolation
// level, another transaction cannot concurrently modify the data that is read
// or modified by this transaction.
func (t *Transaction) Get(key *Key, dst interface{}) error {
err := t.client.get(t.ctx, []*Key{key}, []interface{}{dst}, &pb.ReadOptions{Transaction: t.id})
if me, ok := err.(MultiError); ok {
return me[0]
}
return err
}
// GetMulti is a batch version of Get.
func (t *Transaction) GetMulti(keys []*Key, dst interface{}) error {
if t.id == nil {
return errExpiredTransaction
}
return t.client.get(t.ctx, keys, dst, &pb.ReadOptions{Transaction: t.id})
}
// Put is the transaction-specific version of the package function Put.
//
// Put returns a PendingKey which can be resolved into a Key using the
// return value from a successful Commit. If key is an incomplete key, the
// returned pending key will resolve to a unique key generated by the
// datastore.
func (t *Transaction) Put(key *Key, src interface{}) (*PendingKey, error) {
h, err := t.PutMulti([]*Key{key}, []interface{}{src})
if err != nil {
if me, ok := err.(MultiError); ok {
return nil, me[0]
}
return nil, err
}
return h[0], nil
}
// PutMulti is a batch version of Put. One PendingKey is returned for each
// element of src in the same order.
func (t *Transaction) PutMulti(keys []*Key, src interface{}) ([]*PendingKey, error) {
if t.id == nil {
return nil, errExpiredTransaction
}
mutation, err := putMutation(keys, src)
if err != nil {
return nil, err
}
proto.Merge(t.mutation, mutation)
// Prepare the returned handles, pre-populating where possible.
ret := make([]*PendingKey, len(keys))
for i, key := range keys {
h := &PendingKey{}
if key.Incomplete() {
// This key will be in the final commit result.
t.pending = append(t.pending, h)
} else {
h.key = key
}
ret[i] = h
}
return ret, nil
}
// Delete is the transaction-specific version of the package function Delete.
// Delete enqueues the deletion of the entity for the given key, to be
// committed atomically upon calling Commit.
func (t *Transaction) Delete(key *Key) error {
err := t.DeleteMulti([]*Key{key})
if me, ok := err.(MultiError); ok {
return me[0]
}
return err
}
// DeleteMulti is a batch version of Delete.
func (t *Transaction) DeleteMulti(keys []*Key) error {
if t.id == nil {
return errExpiredTransaction
}
mutation, err := deleteMutation(keys)
if err != nil {
return err
}
proto.Merge(t.mutation, mutation)
return nil
}
// Commit represents the result of a committed transaction.
type Commit struct{}
// Key resolves a pending key handle into a final key.
func (c *Commit) Key(p *PendingKey) *Key {
if c != p.commit {
panic("PendingKey was not created by corresponding transaction")
}
return p.key
}
// PendingKey represents the key for newly-inserted entity. It can be
// resolved into a Key by calling the Key method of Commit.
type PendingKey struct {
key *Key
commit *Commit
}