Method wait

dagql/cache.go:3775–3909

(
	ctx context.Context,
	sessionID string,
	resolver TypeResolver,
	oc *ongoingCall,
	req *CallRequest,
)

}

func (c *Cache) wait(
	ctx context.Context,
	sessionID string,
	resolver TypeResolver,
	oc *ongoingCall,
	req *CallRequest,
) (AnyResult, error) {
	var (
		completionErr error
		canceledErr   error
		completed     bool
	)

	select {
	case <-oc.waitCh:
		completed = true
	case <-ctx.Done():
		canceledErr = context.Cause(ctx)
	}

	if completed {
		completionErr = oc.err
	}

	if !completed {
		c.callsMu.Lock()
		oc.waiters--
		lastWaiter := oc.waiters == 0
		releaseHandoff := lastWaiter && oc.handoffHoldActive
		if lastWaiter {
			delete(c.ongoingCalls, oc.callConcurrencyKeys)
			oc.cancel(canceledErr)
		}
		c.callsMu.Unlock()
		if releaseHandoff && oc.res != nil {
			c.egraphMu.Lock()
			queue, decErr := c.decrementIncomingOwnershipLocked(ctx, oc.res, nil)
			collectReleases, collectErr := c.collectUnownedResultsLocked(context.WithoutCancel(ctx), queue)
			c.egraphMu.Unlock()
			oc.handoffHoldActive = false
			if relErr := errors.Join(decErr, collectErr, runOnReleaseFuncs(context.WithoutCancel(ctx), collectReleases)); relErr != nil {
				return nil, errors.Join(canceledErr, relErr)
			}
		}
		return nil, canceledErr
	}

	if completionErr != nil {
		c.callsMu.Lock()
		oc.waiters--
		lastWaiter := oc.waiters == 0
		if lastWaiter {
			delete(c.ongoingCalls, oc.callConcurrencyKeys)
			oc.cancel(completionErr)
		}
		c.callsMu.Unlock()
		return nil, completionErr
	}