Method waitForLog

kayak/waiter.go:153–185

(ctx context.Context, index uint64)

}

func (r *Runtime) waitForLog(ctx context.Context, index uint64) (l *kt.Log, err error) {
	defer trace.StartRegion(ctx, "waitForLog").End()

	if l, err = r.wal.Get(index); err == nil {
		// exists
		return
	}

	rawItem, _ := r.waitLogMap.LoadOrStore(index, newWaitItem(r, index))
	item := rawItem.(*waitItem)

	if item == nil {
		err = kt.ErrInvalidLog
		return
	}

	item.startFetch()

	select {
	case <-item.waitCh:
		l = item.get()
		if l != nil {
			err = nil
		} else {
			err = errors.Wrapf(kt.ErrInvalidLog, "could not fetch log %d", index)
		}
		r.waitLogMap.Delete(index)
	case <-ctx.Done():
		err = ctx.Err()
	}

	return
}

func (r *Runtime) triggerLogAwaits(l *kt.Log) {
	rawItem, ok := r.waitLogMap.Load(l.Index)