“

Series: S5 Intelligent Decisions · Article 20 | Level: Advanced | Reading Time: 20 min

Function Versioning and A/B Testing

#TL;DR

User-defined function updates cannot be deployed all at once -- version management, canary releases, and A/B testing are needed to control risk. The coomia-dip FunctionVersionManager supports semantic versioning, traffic weight routing, A/B experiment framework, and automatic rollback, ensuring function updates are fully validated before full traffic cutover. This article details the version model, routing strategies, experiment metrics collection, statistical significance testing, and automated release pipelines.

#1. Why Function Version Management

#1.1 Risk of Direct Replacement

Code

Direct Replacement vs Gradual Release:

  Direct replacement:
  v1.0 -----------------> v2.0 (100% switch)
       |                  |
       |                  +-> If v2.0 has bugs -> total failure
       |
  Gradual release:
  v1.0 -------- 90% -------- 80% ---- ... ---- 0%
  v2.0 -------- 10% -------- 20% ---- ... ---- 100%
       |         |             |
       |         +-- Monitor   +-- No issues -> continue
       |             Anomaly -> auto-rollback

#2. Version Model

Python

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class ReleaseStage(Enum):
    CANARY = "canary"
    GRADUAL = "gradual"
    MAJORITY = "majority"
    FULL = "full"
    ROLLBACK = "rollback"


@dataclass
class FunctionVersion:
    function_id: str
    version: str
    stage: ReleaseStage = ReleaseStage.CANARY
    traffic_weight: float = 0.0
    is_default: bool = False
    created_at: datetime = field(default_factory=datetime.utcnow)
    metrics: VersionMetrics | None = None
    changelog: str = ""


@dataclass
class VersionMetrics:
    total_invocations: int = 0
    success_count: int = 0
    error_count: int = 0
    timeout_count: int = 0
    avg_duration_ms: float = 0.0
    p99_duration_ms: float = 0.0
    avg_confidence: float = 0.0


@dataclass
class TrafficRule:
    function_id: str
    rules: list[VersionWeight]
    sticky_key: str | None = None


@dataclass
class VersionWeight:
    version: str
    weight: float
    conditions: dict[str, Any] = field(default_factory=dict)

#3. Traffic Routing

Python

import hashlib
import random


class TrafficRouter:
    """Traffic router"""

    def __init__(self):
        self._rules: dict[str, TrafficRule] = {}

    def set_rule(self, rule: TrafficRule) -> None:
        self._rules[rule.function_id] = rule

    def route(self, function_id: str,
              context: dict[str, Any] | None = None) -> str:
        rule = self._rules.get(function_id)
        if rule is None:
            raise ValueError(f"No traffic rule for {function_id}")

        for vw in rule.rules:
            if vw.conditions and self._match_conditions(vw.conditions, context):
                return vw.version

        if rule.sticky_key and context:
            sticky_value = context.get(rule.sticky_key, "")
            if sticky_value:
                return self._sticky_route(rule, sticky_value)

        return self._weighted_route(rule)

    def _weighted_route(self, rule: TrafficRule) -> str:
        rand = random.random()
        cumulative = 0.0
        for vw in rule.rules:
            cumulative += vw.weight
            if rand < cumulative:
                return vw.version
        return rule.rules[-1].version

    def _sticky_route(self, rule: TrafficRule, sticky_value: str) -> str:
        hash_val = int(hashlib.md5(sticky_value.encode()).hexdigest(), 16) % 10000
        cumulative = 0
        for vw in rule.rules:
            cumulative += int(vw.weight * 10000)
            if hash_val < cumulative:
                return vw.version
        return rule.rules[-1].version

    def _match_conditions(self, conditions: dict, context: dict | None) -> bool:
        if context is None:
            return False
        return all(context.get(k) == v for k, v in conditions.items())

#4. A/B Experiment Framework

#4.1 Experiment Definition

Python

@dataclass
class ABExperiment:
    experiment_id: str
    function_id: str
    control_version: str
    treatment_version: str
    traffic_split: float = 0.1
    primary_metric: str = "success_rate"
    min_sample_size: int = 1000
    confidence_level: float = 0.95
    max_duration_hours: int = 168
    status: str = "running"
    started_at: datetime = field(default_factory=datetime.utcnow)
    result: ExperimentResult | None = None


@dataclass
class ExperimentResult:
    winner: str
    primary_metric_control: float
    primary_metric_treatment: float
    relative_improvement: float
    p_value: float
    is_significant: bool
    confidence_interval: tuple[float, float]
    sample_size_control: int
    sample_size_treatment: int

#4.2 Statistical Testing

Python

import math


class StatisticalTester:
    """Statistical significance testing"""

    @staticmethod
    def two_proportion_z_test(
        successes_a: int, total_a: int,
        successes_b: int, total_b: int,
        confidence_level: float = 0.95,
    ) -> ExperimentResult:
        p_a = successes_a / total_a if total_a > 0 else 0
        p_b = successes_b / total_b if total_b > 0 else 0

        p_pool = (successes_a + successes_b) / (total_a + total_b)
        se = math.sqrt(p_pool * (1 - p_pool) * (1/total_a + 1/total_b))
        z_stat = (p_b - p_a) / se if se > 0 else 0
        p_value = 2 * (1 - StatisticalTester._normal_cdf(abs(z_stat)))

        alpha = 1 - confidence_level
        z_critical = StatisticalTester._z_critical(alpha / 2)
        se_diff = math.sqrt(p_a*(1-p_a)/total_a + p_b*(1-p_b)/total_b)
        ci_lower = (p_b - p_a) - z_critical * se_diff
        ci_upper = (p_b - p_a) + z_critical * se_diff

        relative_improvement = (p_b - p_a) / p_a if p_a > 0 else 0
        is_significant = p_value < (1 - confidence_level)

        winner = "inconclusive"
        if is_significant:
            winner = "treatment" if p_b > p_a else "control"

        return ExperimentResult(
            winner=winner,
            primary_metric_control=p_a,
            primary_metric_treatment=p_b,
            relative_improvement=relative_improvement,
            p_value=p_value,
            is_significant=is_significant,
            confidence_interval=(ci_lower, ci_upper),
            sample_size_control=total_a,
            sample_size_treatment=total_b,
        )

    @staticmethod
    def _normal_cdf(x: float) -> float:
        return 0.5 * (1 + math.erf(x / math.sqrt(2)))

    @staticmethod
    def _z_critical(alpha: float) -> float:
        if alpha <= 0.005: return 2.576
        elif alpha <= 0.01: return 2.326
        elif alpha <= 0.025: return 1.960
        elif alpha <= 0.05: return 1.645
        else: return 1.282

#5. Gradual Release Manager

Python

class GradualReleaseManager:
    """Gradual release management"""

    STAGE_WEIGHTS = {
        ReleaseStage.CANARY: 0.05,
        ReleaseStage.GRADUAL: 0.25,
        ReleaseStage.MAJORITY: 0.75,
        ReleaseStage.FULL: 1.0,
    }

    def __init__(self, router: TrafficRouter, metrics_collector,
                 rollback_threshold: dict[str, float] | None = None):
        self._router = router
        self._metrics = metrics_collector
        self._thresholds = rollback_threshold or {
            "error_rate": 0.05, "p99_latency_ms": 500, "timeout_rate": 0.02,
        }

    async def promote(self, function_id: str,
                       new_version: str, current_version: str) -> dict:
        current_stage = self._get_stage(function_id, new_version)
        metrics = await self._metrics.get_version_metrics(function_id, new_version)
        health = self._check_health(metrics)

        if not health["healthy"]:
            await self._rollback(function_id, new_version, current_version)
            return {"action": "rollback", "reason": health["violations"]}

        next_stage = self._next_stage(current_stage)
        if next_stage is None:
            return {"action": "completed", "stage": "full"}

        weight = self.STAGE_WEIGHTS[next_stage]
        self._router.set_rule(TrafficRule(
            function_id=function_id,
            rules=[
                VersionWeight(new_version, weight),
                VersionWeight(current_version, 1.0 - weight),
            ],
        ))
        return {"action": "promoted", "stage": next_stage.value, "new_weight": weight}

    def _check_health(self, metrics: VersionMetrics) -> dict:
        violations = []
        if metrics.total_invocations > 0:
            error_rate = metrics.error_count / metrics.total_invocations
            if error_rate > self._thresholds["error_rate"]:
                violations.append(f"error_rate {error_rate:.2%} exceeds threshold")
            timeout_rate = metrics.timeout_count / metrics.total_invocations
            if timeout_rate > self._thresholds["timeout_rate"]:
                violations.append(f"timeout_rate {timeout_rate:.2%} exceeds threshold")
        if metrics.p99_duration_ms > self._thresholds["p99_latency_ms"]:
            violations.append(f"p99_latency {metrics.p99_duration_ms}ms exceeds threshold")
        return {"healthy": len(violations) == 0, "violations": violations}

    def _next_stage(self, current: ReleaseStage) -> ReleaseStage | None:
        order = [ReleaseStage.CANARY, ReleaseStage.GRADUAL,
                 ReleaseStage.MAJORITY, ReleaseStage.FULL]
        try:
            idx = order.index(current)
            return order[idx + 1] if idx + 1 < len(order) else None
        except ValueError:
            return ReleaseStage.CANARY

    async def _rollback(self, function_id, bad_version, good_version):
        self._router.set_rule(TrafficRule(
            function_id=function_id,
            rules=[VersionWeight(good_version, 1.0)],
        ))

    def _get_stage(self, function_id, version):
        rule = self._router._rules.get(function_id)
        if rule is None:
            return ReleaseStage.CANARY
        for vw in rule.rules:
            if vw.version == version:
                for stage, w in sorted(self.STAGE_WEIGHTS.items(), key=lambda x: x[1]):
                    if vw.weight <= w:
                        return stage
        return ReleaseStage.CANARY

#6. Release Report

Code

Function Release Report:
===================================

  Function: fn-credit-custom-001
  Version: v1.2.0 -> v2.0.0
  Stage: GRADUAL (25%)
  Duration: 4h 32m

  Metrics Comparison:
  ---------------------------------
  Metric         | v1.2.0  | v2.0.0   | Change
  Success rate   | 98.3%   | 98.7%    | +0.4%
  Avg latency    | 12.5ms  | 8.2ms    | -34.4%
  P99 latency    | 45ms    | 28ms     | -37.8%
  Timeout rate   | 0.3%    | 0.1%     | -66.7%
  Avg confidence | 0.72    | 0.75     | +4.2%

  A/B Test Result:
  ---------------------------------
  Primary metric (success rate): p-value = 0.032 (significant)
  Relative improvement: +0.41%
  95% CI: [+0.05%, +0.77%]
  Sample size: A=8,432  B=2,118
  Conclusion: treatment wins

  Recommendation: All metrics better than old version. Promote to MAJORITY.

#7. gRPC Service

PROTOBUF

syntax = "proto3";
package onto.function.version.v1;

service FunctionVersionService {
    rpc CreateVersion(CreateVersionRequest) returns (CreateVersionResponse);
    rpc Promote(PromoteRequest) returns (PromoteResponse);
    rpc Rollback(RollbackRequest) returns (RollbackResponse);
    rpc GetVersionMetrics(MetricsRequest) returns (MetricsResponse);
    rpc StartExperiment(ExperimentRequest) returns (ExperimentResponse);
    rpc GetExperimentResult(GetResultRequest) returns (ExperimentResultResponse);
    rpc SetTrafficRule(TrafficRuleRequest) returns (TrafficRuleResponse);
}

#8. Practical Example

Python

# Scenario: Credit scoring function v1 -> v2 canary release

# 1. Register new version
new_spec = FunctionSpec(
    function_id="fn-credit-001",
    version="2.0.0",
    source_code="...",
    entry_point="score",
)
await registry.register_function(new_spec)

# 2. Set canary routing (5%)
router.set_rule(TrafficRule(
    function_id="fn-credit-001",
    rules=[
        VersionWeight("2.0.0", 0.05),
        VersionWeight("1.2.0", 0.95),
    ],
    sticky_key="user_id",
))

# 3. Start A/B experiment
experiment = ABExperiment(
    experiment_id="exp-credit-v2",
    function_id="fn-credit-001",
    control_version="1.2.0",
    treatment_version="2.0.0",
    traffic_split=0.05,
    min_sample_size=1000,
)

# 4. Check results after data accumulation
result = tester.two_proportion_z_test(
    successes_a=8302, total_a=8432,
    successes_b=2091, total_b=2118,
)
# ExperimentResult(winner="treatment", p_value=0.032, is_significant=True)

# 5. Promote to next stage
release_result = await release_manager.promote("fn-credit-001", "2.0.0", "1.2.0")
# {"action": "promoted", "stage": "gradual", "new_weight": 0.25}

#Key Takeaways

Semantic versioning standardizes change impact -- major changes require full validation
Weight routing enables precise control over each version's traffic share
Sticky routing ensures the same user always hits the same version for consistent experience
A/B experiments quantify new version effectiveness via two-proportion Z-tests
Gradual release progresses through four stages from 5% canary to 100% full
Auto-rollback switches back to old version when error rate, latency, or timeout rate exceeds thresholds
Release reports auto-generate metric comparisons and statistical test results

#Next Article

Next up: S5-21 generate_bindings: Auto-Generating Type-Safe Function Bindings from Ontology details how coomia-dip auto-generates multi-language Ontology binding code.

tags: #versioning #ab-testing #canary #gradual-release #rollback #traffic-routing #coomia-dip