done signals that the acquiree has completed acquiring a job (usually successfully, but we also get this call if there is a database error other than ErrNoRows). Any error returned is a serious internal error indicating that integrity of the internal state is corrupted by a code bug.
(dk dKey, clearance chan struct{})
| 242 | // there is a database error other than ErrNoRows). Any error returned is a serious internal error indicating that |
| 243 | // integrity of the internal state is corrupted by a code bug. |
| 244 | func (a *Acquirer) done(dk dKey, clearance chan struct{}) error { |
| 245 | a.mu.Lock() |
| 246 | defer a.mu.Unlock() |
| 247 | d, ok := a.q[dk] |
| 248 | if !ok { |
| 249 | // this is a code error, as something removed the domain early, or done |
| 250 | // was called twice. |
| 251 | err := xerrors.New("done for a domain that doesn't exist") |
| 252 | a.logger.Critical(a.ctx, "internal error", slog.Error(err)) |
| 253 | return err |
| 254 | } |
| 255 | w, ok := d.acquirees[clearance] |
| 256 | if !ok { |
| 257 | // this is a code error, as something removed the dKey early, or done |
| 258 | // was called twice. |
| 259 | err := xerrors.New("done for an acquiree that doesn't exist") |
| 260 | a.logger.Critical(a.ctx, "internal error", slog.Error(err)) |
| 261 | return err |
| 262 | } |
| 263 | if !w.inProgress { |
| 264 | err := xerrors.New("done acquiree was not in progress") |
| 265 | a.logger.Critical(a.ctx, "internal error", slog.Error(err)) |
| 266 | return err |
| 267 | } |
| 268 | delete(d.acquirees, clearance) |
| 269 | if len(d.acquirees) == 0 { |
| 270 | d.cancel() |
| 271 | delete(a.q, dk) |
| 272 | return nil |
| 273 | } |
| 274 | // in the mainline, this means that the acquiree successfully got a job. |
| 275 | // if any others are waiting, clear one of them to try to get a job next so |
| 276 | // that we process the jobs until there are no more acquirees or the database |
| 277 | // is empty of jobs meeting our criteria |
| 278 | for _, other := range d.acquirees { |
| 279 | if other.inProgress { |
| 280 | err := xerrors.New("more than one acquiree in progress for same key") |
| 281 | a.logger.Critical(a.ctx, "internal error", slog.Error(err)) |
| 282 | return err |
| 283 | } |
| 284 | other.inProgress = true |
| 285 | other.clearance <- struct{}{} |
| 286 | break // just one |
| 287 | } |
| 288 | return nil |
| 289 | } |
| 290 | |
| 291 | func (a *Acquirer) subscribe() { |
| 292 | subscribed := make(chan struct{}) |