Skip to content

Commit a082c4a

Browse files
pandatixpandatix
authored
Improve etcd-based RW locks and API contextual awarness of cancelation (#1180)
* impr(locks): handle context cancelation with recovery mecanisms * impr(locks): don't return cancelation errors * impr(locks): improve cancelation and error handling with session-awareness * fix(locks): typing and local lock context cancelation * impr(locks): add etcd healthcheck time windows to avoid spamming, add keepalive to avoid idling * impr(api): handle context cancelation for better recovery, fix bugs * fix(etcd): protect the whole healthcheck for time window to avoid duplicating the call (race condition) * docs: add recovery of RWLocks in the design webdoc * fix(api): unprotected segment in delete operation leading to potential race conditions * docs: fix figure and clarify CP-/AP-first decision on using etcd
1 parent 8c384f9 commit a082c4a

18 files changed

Lines changed: 871 additions & 311 deletions

File tree

‎api/v1/challenge/create.go‎

Lines changed: 28 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -35,12 +35,17 @@ func (store *Store) CreateChallenge(ctx context.Context, req *CreateChallengeReq
3535
span.AddEvent("lock TOTW")
3636
totw, err := common.LockTOTW(ctx)
3737
if err != nil {
38+
if totw.IsCanceled(err) {
39+
return nil, nil
40+
}
3841
err := &errs.ErrInternal{Sub: err}
3942
logger.Error(ctx, "build TOTW lock", zap.Error(err))
4043
return nil, errs.ErrInternalNoSub
4144
}
42-
defer common.LClose(totw)
4345
if err := totw.RLock(ctx); err != nil {
46+
if totw.IsCanceled(err) {
47+
return nil, nil
48+
}
4449
err := &errs.ErrInternal{Sub: err}
4550
logger.Error(ctx, "TOTW R lock", zap.Error(err))
4651
return nil, errs.ErrInternalNoSub
@@ -50,31 +55,48 @@ func (store *Store) CreateChallenge(ctx context.Context, req *CreateChallengeReq
5055
// 2. Lock RW challenge
5156
clock, err := common.LockChallenge(ctx, req.Id)
5257
if err != nil {
58+
if clock.IsCanceled(err) {
59+
// If canceled, we need to recover
60+
if err := totw.RUnlock(context.WithoutCancel(ctx)); err != nil {
61+
err := &errs.ErrInternal{Sub: err}
62+
logger.Error(ctx, "recovering from build challenge lock", zap.Error(err))
63+
return nil, errs.ErrInternalNoSub
64+
}
65+
return nil, nil // recovery is successful, we can quit safely
66+
}
5367
err := &errs.ErrInternal{Sub: err}
5468
logger.Error(ctx, "build challenge lock", zap.Error(multierr.Combine(
55-
totw.RUnlock(ctx),
69+
totw.RUnlock(context.WithoutCancel(ctx)),
5670
err,
5771
)))
5872
return nil, errs.ErrInternalNoSub
5973
}
60-
defer common.LClose(clock)
6174
if err := clock.RWLock(ctx); err != nil {
75+
if clock.IsCanceled(err) {
76+
// If canceled, we need to recover
77+
if err := totw.RUnlock(context.WithoutCancel(ctx)); err != nil {
78+
err := &errs.ErrInternal{Sub: err}
79+
logger.Error(ctx, "recovering from challenge RW lock", zap.Error(err))
80+
return nil, errs.ErrInternalNoSub
81+
}
82+
return nil, nil // recovery is successful, we can quit safely
83+
}
6284
err := &errs.ErrInternal{Sub: err}
6385
logger.Error(ctx, "challenge RW lock", zap.Error(multierr.Combine(
64-
totw.RUnlock(ctx),
86+
totw.RUnlock(context.WithoutCancel(ctx)),
6587
err,
6688
)))
6789
return nil, errs.ErrInternalNoSub
6890
}
6991
defer func(lock lock.RWLock) {
70-
if err := lock.RWUnlock(ctx); err != nil {
92+
if err := lock.RWUnlock(context.WithoutCancel(ctx)); err != nil {
7193
err := &errs.ErrInternal{Sub: err}
7294
logger.Error(ctx, "challenge RW unlock", zap.Error(err))
7395
}
7496
}(clock)
7597

7698
// 3. Unlock R TOTW
77-
if err := totw.RUnlock(ctx); err != nil {
99+
if err := totw.RUnlock(context.WithoutCancel(ctx)); err != nil {
78100
err := &errs.ErrInternal{Sub: err}
79101
logger.Error(ctx, "TOTW R unlock", zap.Error(err))
80102
return nil, errs.ErrInternalNoSub

‎api/v1/challenge/delete.go‎

Lines changed: 48 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,17 @@ func (store *Store) DeleteChallenge(ctx context.Context, req *DeleteChallengeReq
2727
span.AddEvent("lock TOTW")
2828
totw, err := common.LockTOTW(ctx)
2929
if err != nil {
30+
if totw.IsCanceled(err) {
31+
return nil, nil
32+
}
3033
err := &errs.ErrInternal{Sub: err}
3134
logger.Error(ctx, "build TOTW lock", zap.Error(err))
3235
return nil, errs.ErrInternalNoSub
3336
}
34-
defer common.LClose(totw)
3537
if err := totw.RLock(ctx); err != nil {
38+
if totw.IsCanceled(err) {
39+
return nil, nil
40+
}
3641
err := &errs.ErrInternal{Sub: err}
3742
logger.Error(ctx, "TOTW R lock", zap.Error(err))
3843
return nil, errs.ErrInternalNoSub
@@ -42,31 +47,51 @@ func (store *Store) DeleteChallenge(ctx context.Context, req *DeleteChallengeReq
4247
// 2. Lock RW challenge
4348
clock, err := common.LockChallenge(ctx, req.Id)
4449
if err != nil {
50+
if clock.IsCanceled(err) {
51+
// If canceled, we need to recover
52+
if err := totw.RUnlock(context.WithoutCancel(ctx)); err != nil {
53+
err := &errs.ErrInternal{Sub: err}
54+
logger.Error(ctx, "recovering from build challenge lock", zap.Error(err))
55+
return nil, errs.ErrInternalNoSub
56+
}
57+
return nil, nil // recovery is successful, we can quit safely
58+
}
4559
err := &errs.ErrInternal{Sub: err}
4660
logger.Error(ctx, "build challenge lock", zap.Error(multierr.Combine(
47-
totw.RUnlock(ctx),
61+
totw.RUnlock(context.WithoutCancel(ctx)),
4862
err,
4963
)))
5064
return nil, errs.ErrInternalNoSub
5165
}
52-
defer common.LClose(clock)
5366
if err := clock.RWLock(ctx); err != nil {
67+
if clock.IsCanceled(err) {
68+
// If canceled, we need to recover
69+
if err := totw.RUnlock(context.WithoutCancel(ctx)); err != nil {
70+
err := &errs.ErrInternal{Sub: err}
71+
logger.Error(ctx, "recovering from challenge RW lock", zap.Error(err))
72+
return nil, errs.ErrScenarioNoSub
73+
}
74+
return nil, nil // recovery is successful, we can quit safely
75+
}
5476
err := &errs.ErrInternal{Sub: err}
5577
logger.Error(ctx, "challenge RW lock", zap.Error(multierr.Combine(
56-
totw.RUnlock(ctx),
78+
totw.RUnlock(context.WithoutCancel(ctx)),
5779
err,
5880
)))
5981
return nil, errs.ErrInternalNoSub
6082
}
83+
defer func(lock lock.RWLock) {
84+
if err := lock.RWUnlock(context.WithoutCancel(ctx)); err != nil {
85+
err := &errs.ErrInternal{Sub: err}
86+
logger.Error(ctx, "challenge RW unlock", zap.Error(err))
87+
}
88+
}(clock)
6189
// don't defer unlock, will do it manually for ASAP challenge availability
6290

6391
// 3. Unlock R TOTW
64-
if err := totw.RUnlock(ctx); err != nil {
92+
if err := totw.RUnlock(context.WithoutCancel(ctx)); err != nil {
6593
err := &errs.ErrInternal{Sub: err}
66-
logger.Error(ctx, "TOTW R unlock", zap.Error(multierr.Combine(
67-
clock.RWUnlock(ctx),
68-
err,
69-
)))
94+
logger.Error(ctx, "TOTW R unlock", zap.Error(err))
7095
return nil, errs.ErrInternalNoSub
7196
}
7297
span.AddEvent("unlocked TOTW")
@@ -75,73 +100,32 @@ func (store *Store) DeleteChallenge(ctx context.Context, req *DeleteChallengeReq
75100
fschall, err := fs.LoadChallenge(req.Id)
76101
if err != nil {
77102
if err, ok := err.(*errs.ErrInternal); ok {
78-
logger.Error(ctx, "reading challenge from filesystem",
79-
zap.Error(multierr.Combine(
80-
clock.RWUnlock(ctx),
81-
err,
82-
)),
83-
)
103+
logger.Error(ctx, "reading challenge from filesystem", zap.Error(err))
84104
return nil, errs.ErrInternalNoSub
85105
}
86-
if err := clock.RWUnlock(ctx); err != nil {
87-
logger.Error(ctx, "reading challenge from filesystem",
88-
zap.Error(clock.RWUnlock(ctx)),
89-
)
90-
}
91106
return nil, err
92107
}
93108

94109
// 5. Create "relock" and "work" wait groups for all instances, and for each
95110
ists, err := fs.ListInstances(req.Id)
96111
if err != nil {
97112
err := &errs.ErrInternal{Sub: err}
98-
logger.Error(ctx, "listing instances",
99-
zap.Error(multierr.Combine(
100-
clock.RWUnlock(ctx),
101-
err,
102-
)),
103-
)
113+
logger.Error(ctx, "listing instances", zap.Error(err))
104114
return nil, errs.ErrInternalNoSub
105115
}
106116

107117
logger.Info(ctx, "deleting challenge",
108118
zap.Int("instances", len(ists)),
109119
)
110-
relock := &sync.WaitGroup{} // track goroutines that overlocked an identity
111-
relock.Add(len(ists))
112120
work := &sync.WaitGroup{} // track goroutines that ended dealing with the instances
113121
work.Add(len(ists))
114122
cerr := make(chan error, len(ists))
115123
for _, identity := range ists {
116-
go func(relock, work *sync.WaitGroup, cerr chan<- error, identity string) {
117-
// 6.e. done in the "work" wait group
124+
go func(work *sync.WaitGroup, cerr chan<- error, identity string) {
125+
// 6.b. done in the "work" wait group
118126
defer work.Done()
119127

120-
// 6.a. Lock RW instance
121-
ilock, err := common.LockInstance(ctx, req.Id, identity)
122-
if err != nil {
123-
cerr <- err
124-
relock.Done() // release to avoid dead-lock
125-
return
126-
}
127-
defer common.LClose(ilock)
128-
if err := ilock.RWLock(ctx); err != nil {
129-
cerr <- err
130-
relock.Done() // release to avoid dead-lock
131-
return
132-
}
133-
defer func(lock lock.RWLock) {
134-
// 6.d. Unlock RW instance
135-
if err := lock.RWUnlock(ctx); err != nil {
136-
err := &errs.ErrInternal{Sub: err}
137-
logger.Error(ctx, "instance RW unlock", zap.Error(err))
138-
}
139-
}(ilock)
140-
141-
// 6.b. done in the "relock" wait group
142-
relock.Done()
143-
144-
// 6.c. delete it
128+
// 6.a. delete it
145129
fsist, err := fs.LoadInstance(req.Id, identity)
146130
if err != nil {
147131
cerr <- err
@@ -162,18 +146,10 @@ func (store *Store) DeleteChallenge(ctx context.Context, req *DeleteChallengeReq
162146
common.InstancesUDCounter().Add(ctx, -1,
163147
metric.WithAttributeSet(common.InstanceAttrs(req.Id, sourceID, sourceID != "")),
164148
)
165-
}(relock, work, cerr, identity)
149+
}(work, cerr, identity)
166150
}
167151

168-
// 7. Once all "relock" done, unlock RW challenge
169-
relock.Wait()
170-
if err := clock.RWUnlock(ctx); err != nil {
171-
err := &errs.ErrInternal{Sub: err}
172-
logger.Error(ctx, "challenge RW unlock", zap.Error(err))
173-
return nil, errs.ErrInternalNoSub
174-
}
175-
176-
// 8. Once all "work" done, return response or error if any
152+
// 7. Once all "work" done, return response or error if any
177153
work.Wait()
178154
close(cerr)
179155
var merri, merr error
@@ -196,16 +172,18 @@ func (store *Store) DeleteChallenge(ctx context.Context, req *DeleteChallengeReq
196172
}
197173
return nil, errs.ErrInternalNoSub
198174
}
199-
if merr != nil {
200-
return nil, merr
201-
}
202-
203175
if err := fschall.Delete(); err != nil {
204176
logger.Error(ctx, "removing challenge directory",
205-
zap.Error(err),
177+
zap.Error(multierr.Combine(
178+
merr, // keep instance processing error(s)
179+
err,
180+
)),
206181
)
207182
return nil, errs.ErrInternalNoSub
208183
}
184+
if merr != nil {
185+
return nil, merr
186+
}
209187

210188
logger.Info(ctx, "challenge deleted successfully")
211189
common.ChallengesUDCounter().Add(ctx, -1)

‎api/v1/challenge/query.go‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package challenge
22

33
import (
4+
"context"
45
"sync"
56

67
"go.opentelemetry.io/otel/attribute"
@@ -27,12 +28,17 @@ func (store *Store) QueryChallenge(_ *emptypb.Empty, server ChallengeStore_Query
2728
span.AddEvent("lock TOTW")
2829
totw, err := common.LockTOTW(ctx)
2930
if err != nil {
31+
if totw.IsCanceled(err) {
32+
return nil
33+
}
3034
err := &errs.ErrInternal{Sub: err}
3135
logger.Error(ctx, "build TOTW lock", zap.Error(err))
3236
return errs.ErrInternalNoSub
3337
}
34-
defer common.LClose(totw)
3538
if err := totw.RWLock(ctx); err != nil {
39+
if totw.IsCanceled(err) {
40+
return nil
41+
}
3642
err := &errs.ErrInternal{Sub: err}
3743
logger.Error(ctx, "TOTW RW lock", zap.Error(err))
3844
return errs.ErrInternalNoSub
@@ -45,7 +51,7 @@ func (store *Store) QueryChallenge(_ *emptypb.Empty, server ChallengeStore_Query
4551
err := &errs.ErrInternal{Sub: err}
4652
logger.Error(ctx, "listing challenges",
4753
zap.Error(multierr.Append(
48-
totw.RWUnlock(ctx),
54+
totw.RWUnlock(context.WithoutCancel(ctx)),
4955
err,
5056
)),
5157
)
@@ -74,19 +80,27 @@ func (store *Store) QueryChallenge(_ *emptypb.Empty, server ChallengeStore_Query
7480
// 4.a. Lock R challenge
7581
clock, err := common.LockChallenge(ctx, id)
7682
if err != nil {
83+
// If the context has been canceled, it is not a big problem, just stops here
84+
if clock.IsCanceled(err) {
85+
err = nil
86+
}
7787
cerr <- err
7888
relock.Done() // release to avoid dead-lock
7989
return
8090
}
81-
defer common.LClose(clock)
8291
if err := clock.RLock(ctx); err != nil {
92+
// If the context has been canceled, it is not a big problem, just stops here
93+
if clock.IsCanceled(err) {
94+
err = nil
95+
}
8396
cerr <- err
8497
relock.Done() // release to avoid dead-lock
8598
return
8699
}
87100
defer func(lock lock.RWLock) {
88101
// 4.e. Unlock R challenge
89-
if err := lock.RUnlock(ctx); err != nil {
102+
// If cancel happens in the meantime, we still need to free the lock
103+
if err := lock.RUnlock(context.WithoutCancel(ctx)); err != nil {
90104
err := &errs.ErrInternal{Sub: err}
91105
logger.Error(ctx, "challenge RW unlock", zap.Error(err))
92106
}
@@ -167,7 +181,7 @@ func (store *Store) QueryChallenge(_ *emptypb.Empty, server ChallengeStore_Query
167181

168182
// 5. Once all "relock" done, unlock RW TOTW
169183
relock.Wait()
170-
if err := totw.RWUnlock(ctx); err != nil {
184+
if err := totw.RWUnlock(context.WithoutCancel(ctx)); err != nil {
171185
err := &errs.ErrInternal{Sub: err}
172186
logger.Error(ctx, "TOTW RW unlock", zap.Error(err))
173187
return errs.ErrInternalNoSub

0 commit comments

Comments
 (0)