# Edge-to-Cloud Data Synchronization

**Last updated:** 2026-02
**Status:** GA
**Category:** Hybrid Cloud & Edge AI

---

## Introduksjon

Palitelig datasynkronisering mellom edge og sky er en av de mest komplekse utfordringene i hybrid AI-arkitekturer. Data ma flyte i begge retninger — sensordata og inferensresultater fra edge til sky for langsiktig analyse og modelltrening, og oppdaterte modeller og konfigurasjoner fra sky tilbake til edge. Alt dette ma handtere nettverksavbrudd, konflikter og dataintegritet.

For norsk offentlig sektor er palitelig synkronisering kritisk: inspeksjonsdata fra felt ma garantert na sentrale systemer, AI-modellsoppdateringer ma distribueres til alle edge-stasjoner uten manuell intervensjon, og logging for compliance-formal ma vaere komplett — selv etter langvarige offline-perioder.

Microsoft tilbyr flere synkroniseringsmekanismer: Azure IoT Edge med utvidet offline-stoette (ubegrenset offline-tid med lokal buffring), Azure Container Storage med automatisk sky-sync, Azure Cosmos DB med multi-region replikering, og Azure IoT Hub device twins for konfigurasjonssynkronisering. Valget avhenger av datamengde, konsistenskrav og tilkoblingsprofil.

---

## Kjernekomponenter

| Komponent | Formal | Teknologi |
|-----------|--------|-----------|
| Azure IoT Edge | Utvidet offline med meldingsbuffring | Container runtime |
| Azure Container Storage | Lokal lagring med automatisk sky-sync | Arc-enabled |
| Azure Cosmos DB | Multi-region database med konflikthondtering | NoSQL / SQL API |
| IoT Hub Device Twins | Konfigurasjonssynkronisering enhet-sky | PaaS |
| Event Hub | Hoeyvolum event-inntak | Event streaming |
| Azure Data Lake | Langsiktig datalagring | Storage Gen2 |
| Delta Lake | ACID-transaksjoner pa datalake | Open source |

---

## Eventual Consistency Patterns

### Konsistensmodeller for edge-sky

| Modell | Beskrivelse | Latens | Brukstilfelle |
|--------|-------------|--------|---------------|
| Sterk konsistens | Alle noder ser samme data samtidig | Hoey | Kritiske transaksjoner |
| Bounded staleness | Data er konsistent innen et tidsvindu | Medium | Nesten-sanntid dashboards |
| Session konsistens | Konsistens innen en enhet-session | Lav | Brukerinteraksjon |
| Eventuell konsistens | Data konvergerer over tid | Lavest | Sensordata, logger |

### Azure Cosmos DB for edge-sky-synkronisering

```python
# Azure Cosmos DB med konfigurerbar konsistens for edge-sky sync
from azure.cosmos import CosmosClient, PartitionKey
from azure.cosmos.documents import ConsistencyLevel

class EdgeCloudSyncStore:
    def __init__(self, endpoint: str, key: str):
        self.client = CosmosClient(
            endpoint, key,
            consistency_level=ConsistencyLevel.Session  # Bra for edge-sky
        )
        self.database = self.client.get_database_client("edge-ai-data")

    def setup_containers(self):
        """Opprett containere for synkronisert data"""

        # Sensordata: Eventuell konsistens, hoey throughput
        self.sensor_container = self.database.create_container_if_not_exists(
            id="sensor-data",
            partition_key=PartitionKey(path="/deviceId"),
            default_ttl=86400 * 30  # 30 dagers retention
        )

        # AI-resultater: Session konsistens
        self.ai_results = self.database.create_container_if_not_exists(
            id="ai-results",
            partition_key=PartitionKey(path="/deviceId"),
            default_ttl=86400 * 365  # 1 aars retention
        )

        # Modellkonfigurasjon: Sterkere konsistens
        self.model_config = self.database.create_container_if_not_exists(
            id="model-config",
            partition_key=PartitionKey(path="/region")
        )

    def upsert_sensor_data(self, device_id: str, readings: list[dict]):
        """Skriv sensordata med idempotensnokkel for a haandtere re-sync"""
        for reading in readings:
            reading["id"] = f"{device_id}_{reading['timestamp']}"
            reading["deviceId"] = device_id
            reading["_etag"] = None  # Cosmos DB haandterer versjonering

            self.sensor_container.upsert_item(
                body=reading,
                pre_trigger_include=None,
                post_trigger_include=None
            )

    def get_latest_model_config(self, region: str) -> dict:
        """Hent siste modellkonfigurasjon for en region"""
        query = """
            SELECT TOP 1 *
            FROM c
            WHERE c.region = @region
            ORDER BY c.updatedAt DESC
        """
        items = list(self.model_config.query_items(
            query=query,
            parameters=[{"name": "@region", "value": region}],
            enable_cross_partition_query=False
        ))
        return items[0] if items else None
```

### Event-basert synkronisering

```python
# Event-basert edge-to-cloud synkronisering
import asyncio
import json
import gzip
from datetime import datetime, timedelta

class EventBasedSync:
    def __init__(self, local_store, cloud_endpoint: str):
        self.local_store = local_store
        self.cloud_endpoint = cloud_endpoint
        self.sync_log = []
        self.last_sync = None

    async def sync_outbound(self, max_batch_size: int = 100) -> dict:
        """Synkroniser lokale hendelser til sky"""
        # Hent usynkroniserte hendelser
        pending = self.local_store.get_unsynced_events(limit=max_batch_size)

        if not pending:
            return {"status": "up_to_date", "synced": 0}

        # Komprimer for overfoering
        payload = gzip.compress(
            json.dumps([e.__dict__ for e in pending]).encode()
        )

        try:
            # Send til sky-endpoint
            response = await self._send_to_cloud(payload)

            if response["status"] == "accepted":
                # Marker som synkronisert
                event_ids = [e.id for e in pending]
                self.local_store.mark_synced(event_ids)

                self.last_sync = datetime.utcnow()
                self.sync_log.append({
                    "direction": "outbound",
                    "events": len(event_ids),
                    "size_bytes": len(payload),
                    "timestamp": self.last_sync.isoformat()
                })

                return {
                    "status": "synced",
                    "synced": len(event_ids),
                    "remaining": self.local_store.count_unsynced(),
                    "compressed_size": len(payload)
                }

        except ConnectionError:
            return {
                "status": "offline",
                "pending": len(pending),
                "retry_after": "next_connectivity"
            }

    async def sync_inbound(self) -> dict:
        """Hent oppdateringer fra sky (modeller, konfigurasjon)"""
        try:
            since = self.last_sync or datetime.utcnow() - timedelta(days=7)
            updates = await self._fetch_from_cloud(since)

            applied = 0
            for update in updates:
                if update["type"] == "model_update":
                    await self._apply_model_update(update)
                    applied += 1
                elif update["type"] == "config_change":
                    await self._apply_config_change(update)
                    applied += 1

            return {"status": "updated", "applied": applied}

        except ConnectionError:
            return {"status": "offline", "using_cached": True}
```

---

## Delta Sync Optimization

### Inkrementell synkronisering

```python
# Delta-synkronisering for effektiv dataoverfoering
import hashlib
import json
from typing import Optional

class DeltaSyncEngine:
    def __init__(self):
        self.local_checksums: dict[str, str] = {}
        self.sync_watermark: Optional[str] = None

    def calculate_delta(self, current_data: dict,
                        last_synced_data: dict) -> dict:
        """Beregn delta mellom navaerende og sist synkronisert tilstand"""
        delta = {
            "added": {},
            "modified": {},
            "deleted": []
        }

        # Finn nye og endrede elementer
        for key, value in current_data.items():
            current_hash = self._hash_value(value)
            if key not in last_synced_data:
                delta["added"][key] = value
            elif self._hash_value(last_synced_data[key]) != current_hash:
                delta["modified"][key] = value

        # Finn slettede elementer
        for key in last_synced_data:
            if key not in current_data:
                delta["deleted"].append(key)

        return delta

    def apply_delta(self, base_data: dict, delta: dict) -> dict:
        """Anvend delta pa basisdatasettet"""
        result = dict(base_data)

        # Legg til nye
        result.update(delta.get("added", {}))

        # Oppdater endrede
        result.update(delta.get("modified", {}))

        # Fjern slettede
        for key in delta.get("deleted", []):
            result.pop(key, None)

        return result

    def compress_delta(self, delta: dict) -> bytes:
        """Komprimer delta for overfoering"""
        import gzip
        json_bytes = json.dumps(delta, separators=(',', ':')).encode()
        compressed = gzip.compress(json_bytes, compresslevel=9)

        return compressed

    def get_sync_stats(self, delta: dict, compressed: bytes) -> dict:
        """Beregn synkroniseringsstatistikk"""
        full_size = len(json.dumps(delta).encode())
        return {
            "items_changed": (
                len(delta.get("added", {})) +
                len(delta.get("modified", {})) +
                len(delta.get("deleted", []))
            ),
            "full_size_bytes": full_size,
            "compressed_size_bytes": len(compressed),
            "compression_ratio": f"{(1 - len(compressed)/full_size)*100:.1f}%"
                if full_size > 0 else "N/A"
        }

    def _hash_value(self, value) -> str:
        return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()[:16]
```

---

## Conflict Resolution Strategies

### Konflikttyper i edge-sky-synkronisering

| Konflikttype | Arsak | Losningsstrategi |
|-------------|-------|-----------------|
| Write-Write | Samme data endret pa bade edge og sky | LWW eller custom merge |
| Delete-Update | Data slettet pa en side, oppdatert pa annen | Policy-basert (behold eller slett) |
| Schema-conflict | Modellversjon ulik pa edge og sky | Versjonert schema med migrasjon |
| Ordering-conflict | Hendelser mottat i feil rekkefolge | Timestamp-basert reordering |

### Cosmos DB konflikthondtering

```python
# Cosmos DB konflikthondterings-policy
from azure.cosmos import ContainerProxy

class CosmosConflictHandler:
    def __init__(self, container: ContainerProxy):
        self.container = container

    def setup_lww_policy(self):
        """Last-Write-Wins basert pa egendefinert felt"""
        # Konfigureres ved container-oppretting
        # Cosmos DB bruker _ts (timestamp) som default
        pass

    def setup_custom_resolution(self):
        """Custom konflikthondtering med stored procedure"""
        sproc_body = """
        function resolve(incomingItem, existingItem, isTombstone, conflictingItems) {
            // For AI-resultater: Behold den med hoeyest confidence
            if (incomingItem.ai_confidence > existingItem.ai_confidence) {
                return incomingItem;
            }
            return existingItem;
        }
        """
        self.container.scripts.create_stored_procedure({
            "id": "resolveConflict",
            "body": sproc_body
        })

    def read_conflict_feed(self) -> list[dict]:
        """Les konflikter som krever manuell losning"""
        conflicts = list(self.container.list_conflicts())
        return [{
            "id": c["id"],
            "resource_id": c.get("resourceId"),
            "conflict_type": c.get("operationType"),
            "source_region": c.get("sourceResourceId")
        } for c in conflicts]
```

---

## Data Deduplication at Scale

### Dedupliseringsstrategier

```python
# Skalerbar deduplisering for edge-sky-data
import hashlib
from bloom_filter2 import BloomFilter

class EdgeDeduplication:
    def __init__(self, expected_items: int = 1_000_000):
        # Bloom-filter for hurtig duplikat-sjekk (minneeffektivt)
        self.bloom = BloomFilter(
            max_elements=expected_items,
            error_rate=0.01  # 1% falsk-positiv rate
        )
        # Eksakt sjekk for bloom-positive
        self.recent_hashes: set = set()
        self.max_recent = 100_000

    def is_duplicate(self, data: dict) -> bool:
        """Sjekk om dataelementet allerede er prosessert"""
        data_hash = self._compute_hash(data)

        # Hurtig bloom-filter-sjekk
        if data_hash not in self.bloom:
            return False

        # Eksakt sjekk for bekreftelse
        return data_hash in self.recent_hashes

    def mark_processed(self, data: dict):
        """Marker dataelement som prosessert"""
        data_hash = self._compute_hash(data)
        self.bloom.add(data_hash)
        self.recent_hashes.add(data_hash)

        # Begrens minnebruk
        if len(self.recent_hashes) > self.max_recent:
            # Fjern eldste 20%
            to_remove = len(self.recent_hashes) - int(self.max_recent * 0.8)
            for _ in range(to_remove):
                self.recent_hashes.pop()

    def _compute_hash(self, data: dict) -> str:
        """Beregn deterministisk hash av dataelementet"""
        # Bruk innholds-hash (ekskluder metadata som timestamp)
        content_keys = sorted(k for k in data.keys()
                             if k not in ("_ts", "synced_at", "sync_id"))
        content = {k: data[k] for k in content_keys}
        return hashlib.sha256(
            json.dumps(content, sort_keys=True).encode()
        ).hexdigest()

    def get_stats(self) -> dict:
        return {
            "bloom_filter_items": len(self.bloom),
            "recent_exact_items": len(self.recent_hashes),
            "estimated_memory_mb": (
                self.bloom.bitarray.nbytes / 1024 / 1024 +
                len(self.recent_hashes) * 64 / 1024 / 1024
            )
        }
```

---

## Norsk offentlig sektor

### Synkroniseringskrav for offentlig sektor

| Krav | Beskrivelse | Losning |
|------|-------------|---------|
| Dataintegritet | Ingen datatap ved offline/sync | Event sourcing + idempotens |
| Sporbarhet | All synkronisering ma logges | Sync audit log |
| Personvern | Sensitive data ma krypteres i transit | TLS 1.3 + end-to-end |
| Compliance | 7 ars retention for visse datatyper | Immutable storage |
| Konflikthondtering | Sporbar og deterministisk | Policy-basert med audit trail |

### Anbefalte Azure-tjenester per scenario

| Scenario | Primaer-tjeneste | Sekundaer | Konsistens |
|----------|-----------------|-----------|------------|
| IoT-sensordata | IoT Hub + Event Hub | Data Lake | Eventuell |
| AI-resultater | Cosmos DB | Data Lake backup | Session |
| Modellkonfig | IoT Hub Device Twin | Git (GitOps) | Sterk |
| Inspeksjonsdata | Cosmos DB | Blob Storage | Bounded staleness |

---

## Beslutningsrammeverk

| Scenario | Anbefaling | Begrunnelse |
|----------|------------|-------------|
| Hoeyvolum sensorer, enveis | IoT Hub → Event Hub → Data Lake | Skalerbart, rimelig |
| Toveis med konfliktfare | Cosmos DB med session-konsistens | Innebygd konflikthondtering |
| Kritisk data, null tap | Event sourcing + Cosmos DB | Idempotent, sporbar |
| Periodisk bulk-sync | Delta sync + Azure Blob | Minimal bandwidth |
| Multi-edge koordinering | Cosmos DB multi-write | Automatisk konflikthondtering |
| Modellpush til edge | IoT Hub Device Twin + Blob SAS | Etablert monster |

---

## For Cosmo

- **Event sourcing med idempotens er gullstandarden** for edge-sky-synkronisering — alle dataelementer faar en unik ID og kan trygt re-sendes uten duplikater
- **Delta-synkronisering reduserer datavolum med 80-95%** sammenlignet med full sync — beregn kun endringer og komprimer med gzip for minimal bandbreddebruk
- **Cosmos DB med session-konsistens er den beste balansen** mellom ytelse og dataintegritet for de fleste edge-sky-scenarier i offentlig sektor
- **Bloom-filter gir O(1) dedupliseringssjekk** med minimal minnebruk — implementer dette pa bade edge og sky-siden for a hindre duplikat-inntak
- **For norsk offentlig sektor: Krav til sporbarhet og retention betyr at ALL synkronisering ma logges** — implementer sync audit log med 7 ars immutable retention for compliance med arkivloven