| 1 | package abono |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync" |
| 6 | ) |
| 7 | |
| 8 | // subLocks serialises the work on one Stripe subscription across every |
| 9 | // entry point that fetches it and writes what it read: the webhook, |
| 10 | // Finish, Reconcile and Adopt (ruling R35). Each holds the subscription's |
| 11 | // lock from before its fetch until its write has committed, so the |
| 12 | // snapshots are written in the order they were read and an older one can |
| 13 | // never overwrite a newer one. SQLite's single writer orders the writes, |
| 14 | // not the fetches, which is why the database alone is not enough. |
| 15 | // |
| 16 | // The lock is in this process: a rastrillo app is one process on one |
| 17 | // SQLite file. Several processes writing one database would need a check |
| 18 | // in the database instead. |
| 19 | type subLocks struct { |
| 20 | mu sync.Mutex |
| 21 | held map[string]*subLock |
| 22 | } |
| 23 | |
| 24 | type subLock struct { |
| 25 | ch chan struct{} // a token while held |
| 26 | users int // holders and waiters, so an idle entry is dropped |
| 27 | } |
| 28 | |
| 29 | // lock waits for id's lock and returns the function that releases it, or |
| 30 | // ctx's error if ctx ends first. |
| 31 | func (l *subLocks) lock(ctx context.Context, id string) (unlock func(), err error) { |
| 32 | l.mu.Lock() |
| 33 | if l.held == nil { |
| 34 | l.held = map[string]*subLock{} |
| 35 | } |
| 36 | e := l.held[id] |
| 37 | if e == nil { |
| 38 | e = &subLock{ch: make(chan struct{}, 1)} |
| 39 | l.held[id] = e |
| 40 | } |
| 41 | e.users++ |
| 42 | l.mu.Unlock() |
| 43 | |
| 44 | done := func() { |
| 45 | l.mu.Lock() |
| 46 | if e.users--; e.users == 0 { |
| 47 | delete(l.held, id) |
| 48 | } |
| 49 | l.mu.Unlock() |
| 50 | } |
| 51 | select { |
| 52 | case e.ch <- struct{}{}: |
| 53 | return func() { <-e.ch; done() }, nil |
| 54 | case <-ctx.Done(): |
| 55 | done() |
| 56 | return nil, ctx.Err() |
| 57 | } |
| 58 | } |
| 59 | |
amadan