nodes_distribute

var DistributeMap = action.New("distribute.map", func(ctx context.Context, req DistributeMapReq) (DistributeMapRes, error) {
	if req.Action == "" {
		return DistributeMapRes{}, xerr.BadRequest("distribute.map: action is required")
	}
	if len(req.Items) == 0 {
		return DistributeMapRes{}, nil
	}
	if len(req.Items) > maxItems {
		return DistributeMapRes{}, xerr.BadRequest("distribute.map: items exceeds limit")
	}

	resolver := contracts.ActionResolverFromContext(ctx)
	if resolver == nil {
		return DistributeMapRes{}, xerr.Internal("distribute.map: no action resolver in context")
	}

	target, ok := resolver.Action(req.Action)
	if !ok {
		return DistributeMapRes{}, xerr.NotFound("distribute.map: action not found: " + req.Action)
	}

	concurrency := req.Concurrency
	if concurrency <= 0 {
		concurrency = defaultConcurrency
	}
	if concurrency > maxConcurrency {
		concurrency = maxConcurrency
	}

	items := make([]Item, len(req.Items))
	group, groupCtx := errgroup.WithContext(ctx)
	group.SetLimit(concurrency)

	for i, value := range req.Items {
		index, val := i, value
		group.Go(func() error {

			if err := groupCtx.Err(); err != nil {
				items[index] = Item{Error: err.Error()}
				return err
			}

			output, err := invokeSafely(groupCtx, target, val)
			if err != nil {
				items[index] = Item{Error: err.Error()}
				return nil
			}

			items[index] = Item{OK: true, Result: output}
			return nil
		})
	}

	if err := group.Wait(); err != nil {
		return DistributeMapRes{}, err
	}

	result := DistributeMapRes{Items: items}
	for i := range items {
		if items[i].OK {
			result.Succeeded++
		} else {
			result.Failed++
		}
	}
	return result, nil
}).Description("Fan out one action over many items with bounded concurrency").
	Tag("distribute", "fanout").
	Build()

var DistributeReduce = action.New("distribute.reduce", func(_ context.Context, req DistributeReduceReq) (DistributeReduceRes, error) {
	return runReduce(req)
}).Description("Fold distribute.map output into a single value").
	Tag("distribute", "fold").
	Build()