Part 6: Building a Mini-Allocation Engine in Go (Production Prototype)

syntax = "proto3";

package allocation.v1;

option go_package = "github.com/tanhdev/allocation/v1;allocationv1";

service AllocationService {
  rpc AllocateOrder (AllocateOrderRequest) returns (AllocateOrderResponse);
  rpc BatchAllocateOrders (BatchAllocateOrdersRequest) returns (BatchAllocateOrdersResponse);
}

message LineItem {
  string sku = 1;
  int32 quantity = 2;
  double unit_price = 3;
}

message CustomerLocation {
  string zip_code = 1;
  double latitude = 2;
  double longitude = 3;
}

message AllocateOrderRequest {
  string order_id = 1;
  CustomerLocation destination = 2;
  repeated LineItem items = 3;
  string delivery_sla = 4; // e.g., "NEXT_DAY", "STANDARD"
  int64 deadline_epoch_ms = 5;
}

message ShipmentDirective {
  string facility_id = 1;
  map<string, int32> allocated_items = 2; // SKU -> Qty
  double estimated_freight_cost = 3;
  string carrier_code = 4;
}

message AllocateOrderResponse {
  string order_id = 1;
  repeated ShipmentDirective shipments = 2;
  int32 total_splits = 3;
  double total_calculated_cost = 4;
  int64 solve_duration_ms = 5;
  bool is_heuristic_fallback = 6;
}

message BatchAllocateOrdersRequest {
  repeated AllocateOrderRequest orders = 1;
}

message BatchAllocateOrdersResponse {
  repeated AllocateOrderResponse results = 1;
}

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"math"
	"sync"
	"sync/atomic"
	"time"
)

// LineItem defines an item requested in an order.
type LineItem struct {
	SKU      string
	Quantity int32
}

// FacilityStock represents available inventory at a warehouse node.
type FacilityStock struct {
	FacilityID string
	Stock      map[string]int32
	UnitFreight map[string]float64
	SplitPenalty float64
	AvailableLabor int32
}

// AllocationDirective encapsulates the dispatch instruction for a single warehouse.
type AllocationDirective struct {
	FacilityID     string
	AllocatedSKUs  map[string]int32
	FreightCost    float64
}

// AllocationResult represents the completed fulfillment plan.
type AllocationResult struct {
	OrderID         string
	Shipments       []AllocationDirective
	TotalCost       float64
	Splits          int32
	Duration        time.Duration
	UsedFallback    bool
}

// AllocationTask encapsulates an in-flight solver request.
type AllocationTask struct {
	Ctx        context.Context
	OrderID    string
	Items      []LineItem
	Facilities []FacilityStock
	ResultChan chan AllocationResult
	ErrChan    chan error
}

// AllocationEngine manages worker pools and optimization execution.
type AllocationEngine struct {
	workerCount int
	taskQueue   chan AllocationTask
	timeout     time.Duration
	isClosed    atomic.Bool
	wg          sync.WaitGroup

	// Metrics counters
	totalRequests  atomic.Uint64
	fallbackCount  atomic.Uint64
	solverFailures atomic.Uint64
}

// NewAllocationEngine initializes the engine with a dedicated worker pool.
func NewAllocationEngine(workerCount int, queueCapacity int, timeout time.Duration) *AllocationEngine {
	engine := &AllocationEngine{
		workerCount: workerCount,
		taskQueue:   make(chan AllocationTask, queueCapacity),
		timeout:     timeout,
	}

	engine.startWorkers()
	return engine
}

// startWorkers spawns isolated worker goroutines.
func (e *AllocationEngine) startWorkers() {
	for i := 0; i < e.workerCount; i++ {
		e.wg.Add(1)
		go func(workerID int) {
			defer e.wg.Done()
			for task := range e.taskQueue {
				e.processTask(workerID, task)
			}
		}(i)
	}
}

// processTask executes the mathematical solver with timeout and fallback.
func (e *AllocationEngine) processTask(workerID int, task AllocationTask) {
	start := time.Now()
	e.totalRequests.Add(1)

	ctx, cancel := context.WithTimeout(task.Ctx, e.timeout)
	defer cancel()

	// Attempt exact mathematical solve
	res, err := e.solveMILP(ctx, task.OrderID, task.Items, task.Facilities)
	if err == nil {
		res.Duration = time.Since(start)
		res.UsedFallback = false
		task.ResultChan <- res
		return
	}

	// Tripped timeout or solver error -> execute deterministic fallback
	e.solverFailures.Add(1)
	e.fallbackCount.Add(1)

	fallbackRes := e.solveDeterministicHeuristic(task.OrderID, task.Items, task.Facilities)
	fallbackRes.Duration = time.Since(start)
	fallbackRes.UsedFallback = true
	task.ResultChan <- fallbackRes
}

// solveMILP simulates native CGo invocation to HiGHS or Google OR-Tools.
func (e *AllocationEngine) solveMILP(
	ctx context.Context,
	orderID string,
	items []LineItem,
	facilities []FacilityStock,
) (AllocationResult, error) {
	// In production, this invokes CGo bindings:
	// C.SolveHighsModel(...)
	select {
	case <-ctx.Done():
		return AllocationResult{}, ctx.Err()
	default:
	}

	// Evaluate if single-warehouse fulfillment is possible
	for _, fac := range facilities {
		hasAll := true
		for _, it := range items {
			if fac.Stock[it.SKU] < it.Quantity {
				hasAll = false
				break
			}
		}

		if hasAll {
			// Found 100% consolidated single-node fulfillment
			allocated := make(map[string]int32)
			var freight float64
			for _, it := range items {
				allocated[it.SKU] = it.Quantity
				freight += fac.UnitFreight[it.SKU] * float64(it.Quantity)
			}

			return AllocationResult{
				OrderID: orderID,
				Shipments: []AllocationDirective{
					{
						FacilityID:    fac.FacilityID,
						AllocatedSKUs: allocated,
						FreightCost:   freight + fac.SplitPenalty,
					},
				},
				TotalCost: freight + fac.SplitPenalty,
				Splits:    1,
			}, nil
		}
	}

	// If complex split is required and takes > timeout, trigger fallback
	return AllocationResult{}, errors.New("multi-facility solver deadline exceeded")
}

// solveDeterministicHeuristic executes a guaranteed sub-millisecond greedy allocation.
func (e *AllocationEngine) solveDeterministicHeuristic(
	orderID string,
	items []LineItem,
	facilities []FacilityStock,
) AllocationResult {
	remaining := make(map[string]int32)
	for _, it := range items {
		remaining[it.SKU] = it.Quantity
	}

	var directives []AllocationDirective
	var totalCost float64

	for {
		// Check if all items satisfied
		done := true
		for _, q := range remaining {
			if q > 0 {
				done = false
				break
			}
		}
		if done {
			break
		}

		// Pick facility satisfying the most remaining demand
		bestFacIdx := -1
		var bestScore int32 = -1

		for idx, fac := range facilities {
			var score int32
			for sku, need := range remaining {
				if need > 0 && fac.Stock[sku] > 0 {
					score += int32(math.Min(float64(need), float64(fac.Stock[sku])))
				}
			}
			if score > bestScore {
				bestScore = score
				bestFacIdx = idx
			}
		}

		if bestFacIdx == -1 || bestScore <= 0 {
			break // Network inventory completely exhausted
		}

		targetFac := &facilities[bestFacIdx]
		assigned := make(map[string]int32)
		var facFreight float64

		for sku, need := range remaining {
			if need > 0 && targetFac.Stock[sku] > 0 {
				allocQty := int32(math.Min(float64(need), float64(targetFac.Stock[sku])))
				assigned[sku] = allocQty
				targetFac.Stock[sku] -= allocQty
				remaining[sku] -= allocQty
				facFreight += targetFac.UnitFreight[sku] * float64(allocQty)
			}
		}

		directives = append(directives, AllocationDirective{
			FacilityID:    targetFac.FacilityID,
			AllocatedSKUs: assigned,
			FreightCost:   facFreight + targetFac.SplitPenalty,
		})
		totalCost += facFreight + targetFac.SplitPenalty
	}

	return AllocationResult{
		OrderID:   orderID,
		Shipments: directives,
		TotalCost: totalCost,
		Splits:    int32(len(directives)),
	}
}

// Allocate schedules an order allocation task onto the worker pool.
func (e *AllocationEngine) Allocate(
	ctx context.Context,
	orderID string,
	items []LineItem,
	facilities []FacilityStock,
) (AllocationResult, error) {
	if e.isClosed.Load() {
		return AllocationResult{}, errors.New("allocation engine is shutting down")
	}

	task := AllocationTask{
		Ctx:        ctx,
		OrderID:    orderID,
		Items:      items,
		Facilities: facilities,
		ResultChan: make(chan AllocationResult, 1),
		ErrChan:    make(chan error, 1),
	}

	// Non-blocking channel push with immediate heuristic fallback if saturated
	select {
	case e.taskQueue <- task:
		select {
		case res := <-task.ResultChan:
			return res, nil
		case err := <-task.ErrChan:
			return AllocationResult{}, err
		case <-ctx.Done():
			return AllocationResult{}, ctx.Err()
		}
	default:
		// Queue saturated -> directly execute fast heuristic in caller goroutine
		e.fallbackCount.Add(1)
		res := e.solveDeterministicHeuristic(orderID, items, facilities)
		res.UsedFallback = true
		return res, nil
	}
}

// Close gracefully drains the worker pool.
func (e *AllocationEngine) Close() {
	if e.isClosed.CompareAndSwap(false, true) {
		close(e.taskQueue)
		e.wg.Wait()
	}
}

func main() {
	engine := NewAllocationEngine(8, 1000, 35*time.Millisecond)
	defer engine.Close()

	ctx := context.Background()
	items := []LineItem{
		{SKU: "SKU-APPLE-MACBOOK", Quantity: 1},
		{SKU: "SKU-USB-C-CABLE", Quantity: 2},
	}

	facilities := []FacilityStock{
		{
			FacilityID:   "FC-DALLAS",
			Stock:        map[string]int32{"SKU-APPLE-MACBOOK": 1, "SKU-USB-C-CABLE": 0},
			UnitFreight:  map[string]float64{"SKU-APPLE-MACBOOK": 5.0, "SKU-USB-C-CABLE": 1.0},
			SplitPenalty: 6.50,
		},
		{
			FacilityID:   "FC-ATLANTA",
			Stock:        map[string]int32{"SKU-APPLE-MACBOOK": 0, "SKU-USB-C-CABLE": 5},
			UnitFreight:  map[string]float64{"SKU-APPLE-MACBOOK": 6.0, "SKU-USB-C-CABLE": 1.2},
			SplitPenalty: 6.50,
		},
	}

	result, err := engine.Allocate(ctx, "ORD-998822", items, facilities)
	if err != nil {
		log.Fatalf("Allocation failed: %v", err)
	}

	fmt.Printf("Order: %s | Splits: %d | Total Cost: $%.2f | Solved in: %v | Fallback: %v\n",
		result.OrderID, result.Splits, result.TotalCost, result.Duration, result.UsedFallback)
}

package main

import (
	"context"
	"testing"
	"time"
)

// TestSingleWarehouseConsolidation verifies 100% consolidation when a single node stocks all items.
func TestSingleWarehouseConsolidation(t *testing.T) {
	engine := NewAllocationEngine(4, 100, 50*time.Millisecond)
	defer engine.Close()

	ctx := context.Background()
	items := []LineItem{
		{SKU: "SKU-MONITOR-4K", Quantity: 1},
		{SKU: "SKU-HDMI-CABLE", Quantity: 2},
	}

	facilities := []FacilityStock{
		{
			FacilityID:   "FC-WEST",
			Stock:        map[string]int32{"SKU-MONITOR-4K": 5, "SKU-HDMI-CABLE": 10},
			UnitFreight:  map[string]float64{"SKU-MONITOR-4K": 8.0, "SKU-HDMI-CABLE": 1.0},
			SplitPenalty: 6.0,
		},
		{
			FacilityID:   "FC-EAST",
			Stock:        map[string]int32{"SKU-MONITOR-4K": 0, "SKU-HDMI-CABLE": 15},
			UnitFreight:  map[string]float64{"SKU-MONITOR-4K": 12.0, "SKU-HDMI-CABLE": 1.5},
			SplitPenalty: 6.0,
		},
	}

	res, err := engine.Allocate(ctx, "ORDER-TEST-001", items, facilities)
	if err != nil {
		t.Fatalf("unexpected allocation error: %v", err)
	}

	if res.Splits != 1 {
		t.Errorf("expected 1 split (consolidated), got %d splits", res.Splits)
	}

	if res.Shipments[0].FacilityID != "FC-WEST" {
		t.Errorf("expected FC-WEST to fulfill, got %s", res.Shipments[0].FacilityID)
	}

	expectedCost := (8.0 * 1) + (1.0 * 2) + 6.0 // 16.0
	if res.TotalCost != expectedCost {
		t.Errorf("expected total cost $%.2f, got $%.2f", expectedCost, res.TotalCost)
	}
}

// TestMultiFacilitySplit verifies greedy fallback when no single warehouse has complete stock.
func TestMultiFacilitySplit(t *testing.T) {
	engine := NewAllocationEngine(4, 100, 50*time.Millisecond)
	defer engine.Close()

	ctx := context.Background()
	items := []LineItem{
		{SKU: "SKU-DESK", Quantity: 1},
		{SKU: "SKU-CHAIR", Quantity: 1},
	}

	facilities := []FacilityStock{
		{
			FacilityID:   "FC-NORTH",
			Stock:        map[string]int32{"SKU-DESK": 1, "SKU-CHAIR": 0},
			UnitFreight:  map[string]float64{"SKU-DESK": 15.0, "SKU-CHAIR": 0.0},
			SplitPenalty: 7.0,
		},
		{
			FacilityID:   "FC-SOUTH",
			Stock:        map[string]int32{"SKU-DESK": 0, "SKU-CHAIR": 1},
			UnitFreight:  map[string]float64{"SKU-DESK": 0.0, "SKU-CHAIR": 12.0},
			SplitPenalty: 7.0,
		},
	}

	res, err := engine.Allocate(ctx, "ORDER-TEST-002", items, facilities)
	if err != nil {
		t.Fatalf("unexpected allocation error: %v", err)
	}

	if res.Splits != 2 {
		t.Errorf("expected 2 split shipments, got %d", res.Splits)
	}

	expectedCost := 15.0 + 7.0 + 12.0 + 7.0 // 41.0
	if res.TotalCost != expectedCost {
		t.Errorf("expected total cost $%.2f, got $%.2f", expectedCost, res.TotalCost)
	}
}

// BenchmarkAllocationThroughput evaluates concurrent throughput across 10,000 requests.
func BenchmarkAllocationThroughput(b *testing.B) {
	engine := NewAllocationEngine(8, 2000, 20*time.Millisecond)
	defer engine.Close()

	ctx := context.Background()
	items := []LineItem{
		{SKU: "SKU-A", Quantity: 1},
		{SKU: "SKU-B", Quantity: 2},
	}
	facilities := []FacilityStock{
		{
			FacilityID:   "FC-1",
			Stock:        map[string]int32{"SKU-A": 100, "SKU-B": 100},
			UnitFreight:  map[string]float64{"SKU-A": 2.0, "SKU-B": 1.0},
			SplitPenalty: 5.0,
		},
	}

	b.ResetTimer()
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			_, err := engine.Allocate(ctx, "BENCH-ORDER", items, facilities)
			if err != nil {
				b.Errorf("benchmark allocation failed: %v", err)
			}
		}
	})
}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: allocation-service
  namespace: logistics
spec:
  replicas: 12
  selector:
    matchLabels:
      app: allocation-service
  template:
    metadata:
      labels:
        app: allocation-service
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9090"
        prometheus.io/path: "/metrics"
    spec:
      containers:
      - name: engine
        image: tanhdev/order-allocation-engine:2027.1
        env:
        - name: GOMAXPROCS
          value: "8"
        - name: SOLVER_TIMEOUT_MS
          value: "35"
        - name: WORKER_POOL_SIZE
          value: "8"
        resources:
          requests:
            cpu: "4000m"
            memory: "4Gi"
          limits:
            cpu: "8000m"
            memory: "8Gi"
        ports:
        - containerPort: 50051
          name: grpc
        - containerPort: 9090
          name: metrics
        livenessProbe:
          grpc:
            port: 50051
          initialDelaySeconds: 5
          periodSeconds: 10
        readinessProbe:
          grpc:
            port: 50051
          initialDelaySeconds: 2
          periodSeconds: 5