CRDT 入门：不靠共识也能合并——但代价是什么

package main

import (
    "fmt"
    "sync"
)

// GCounter 是一个 Grow-only Counter（只增计数器）
// 每个节点只递增自己的槽位，合并时逐位取 max
type GCounter struct {
    mu     sync.Mutex
    nodeID int
    counts []int
}

// NewGCounter 创建一个 G-Counter，nodeID 是当前节点编号，n 是总节点数
func NewGCounter(nodeID, n int) *GCounter {
    return &GCounter{
        nodeID: nodeID,
        counts: make([]int, n),
    }
}

// Increment 当前节点递增 1
func (g *GCounter) Increment() {
    g.mu.Lock()
    defer g.mu.Unlock()
    g.counts[g.nodeID]++
}

// Value 返回计数器的总值
func (g *GCounter) Value() int {
    g.mu.Lock()
    defer g.mu.Unlock()
    sum := 0
    for _, c := range g.counts {
        sum += c
    }
    return sum
}

// Merge 合并另一个副本的状态，逐位取 max
func (g *GCounter) Merge(other *GCounter) {
    g.mu.Lock()
    defer g.mu.Unlock()
    for i := range g.counts {
        if other.counts[i] > g.counts[i] {
            g.counts[i] = other.counts[i]
        }
    }
}

func main() {
    // 三个节点，各自独立递增
    a := NewGCounter(0, 3)
    b := NewGCounter(1, 3)
    c := NewGCounter(2, 3)

    // 节点 A 递增 3 次
    a.Increment()
    a.Increment()
    a.Increment()

    // 节点 B 递增 2 次，并且之前同步过 A 的一次递增
    b.counts[0] = 1 // 模拟之前从 A 同步过一次
    b.Increment()
    b.Increment()

    // 节点 C 递增 1 次，之前各同步过 A 和 B 一次
    c.counts[0] = 1
    c.counts[1] = 1
    c.Increment()

    fmt.Printf("合并前: A=%v B=%v C=%v\n", a.counts, b.counts, c.counts)

    // 合并所有副本到 A
    a.Merge(b)
    a.Merge(c)

    fmt.Printf("合并后: %v\n", a.counts)
    fmt.Printf("总计数: %d\n", a.Value()) // 应该是 6
}

package main

import "fmt"

// PNCounter 用两个 G-Counter 的差来表达"可增可减"
type PNCounter struct {
    P *GCounter // 增量
    N *GCounter // 减量
}

// NewPNCounter 创建一个 PN-Counter
func NewPNCounter(nodeID, n int) *PNCounter {
    return &PNCounter{
        P: NewGCounter(nodeID, n),
        N: NewGCounter(nodeID, n),
    }
}

// Inc 递增
func (pn *PNCounter) Inc() {
    pn.P.Increment()
}

// Dec 递减
func (pn *PNCounter) Dec() {
    pn.N.Increment()
}

// Value 返回当前值 = P 总计 - N 总计
func (pn *PNCounter) Value() int {
    return pn.P.Value() - pn.N.Value()
}

// Merge 合并另一个副本：分别合并 P 和 N
func (pn *PNCounter) Merge(other *PNCounter) {
    pn.P.Merge(other.P)
    pn.N.Merge(other.N)
}

func main() {
    a := NewPNCounter(0, 3)
    b := NewPNCounter(1, 3)

    a.Inc() // +1
    a.Inc() // +1
    a.Dec() // -1  → A = 1

    b.Inc() // +1
    b.Dec() // -1
    b.Dec() // -1  → B = -1

    fmt.Printf("合并前: A=%d, B=%d\n", a.Value(), b.Value())

    a.Merge(b)
    fmt.Printf("合并后: A=%d\n", a.Value()) // 应该是 0 (3 inc - 3 dec)
}

package main

import (
    "fmt"
    "sync"
    "time"
)

// LWWRegister 是一个 Last-Writer-Wins 寄存器
// 合并策略：保留时间戳更大的值
type LWWRegister struct {
    mu        sync.Mutex
    value     string
    timestamp int64 // 纳秒级时间戳
}

// Set 设置值，自动记录当前时间戳
func (r *LWWRegister) Set(value string) {
    r.mu.Lock()
    defer r.mu.Unlock()
    r.value = value
    r.timestamp = time.Now().UnixNano()
}

// SetWithTimestamp 手动指定时间戳（用于测试和同步）
func (r *LWWRegister) SetWithTimestamp(value string, ts int64) {
    r.mu.Lock()
    defer r.mu.Unlock()
    r.value = value
    r.timestamp = ts
}

// Get 获取当前值
func (r *LWWRegister) Get() (string, int64) {
    r.mu.Lock()
    defer r.mu.Unlock()
    return r.value, r.timestamp
}

// Merge 合并另一个副本，保留时间戳更大的
func (r *LWWRegister) Merge(other *LWWRegister) {
    r.mu.Lock()
    defer r.mu.Unlock()
    if other.timestamp > r.timestamp {
        r.value = other.value
        r.timestamp = other.timestamp
    }
}

func main() {
    a := &LWWRegister{}
    b := &LWWRegister{}

    // 节点 A 先写 "hello"（时间戳 100）
    a.SetWithTimestamp("hello", 100)
    // 节点 B 后写 "world"（时间戳 200）
    b.SetWithTimestamp("world", 200)

    // 合并：时间戳大的胜出
    a.Merge(b)
    val, ts := a.Get()
    fmt.Printf("合并结果: value=%q, timestamp=%d\n", val, ts)
    // 输出: 合并结果: value="world", timestamp=200
}

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

// 全局唯一标签生成器
var tagCounter int64

func newTag() string {
    id := atomic.AddInt64(&tagCounter, 1)
    return fmt.Sprintf("tag-%d", id)
}

// ORSet 是一个 Observed-Remove Set
// 每个元素关联一组唯一标签，删除只移除已观察到的标签
// tombstones 记录已被删除的标签，合并时用于阻止被删标签复活
type ORSet struct {
    mu         sync.Mutex
    elements   map[string]map[string]bool // element -> set of tags
    tombstones map[string]bool            // 已删除的 tag 集合
}

// NewORSet 创建一个新的 OR-Set
func NewORSet() *ORSet {
    return &ORSet{
        elements:   make(map[string]map[string]bool),
        tombstones: make(map[string]bool),
    }
}

// Add 添加一个元素，生成唯一标签
func (s *ORSet) Add(elem string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    if s.elements[elem] == nil {
        s.elements[elem] = make(map[string]bool)
    }
    s.elements[elem][newTag()] = true
}

// Remove 删除一个元素：移除当前观察到的所有标签，并记录到 tombstones
// 如果其他节点同时添加了新标签，那些标签不受影响
func (s *ORSet) Remove(elem string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    for tag := range s.elements[elem] {
        s.tombstones[tag] = true
    }
    delete(s.elements, elem)
}

// Contains 查询元素是否存在（至少有一个活跃标签）
func (s *ORSet) Contains(elem string) bool {
    s.mu.Lock()
    defer s.mu.Unlock()
    return len(s.elements[elem]) > 0
}

// Values 返回集合中所有元素
func (s *ORSet) Values() []string {
    s.mu.Lock()
    defer s.mu.Unlock()
    var result []string
    for elem, tags := range s.elements {
        if len(tags) > 0 {
            result = append(result, elem)
        }
    }
    return result
}

// Merge 合并另一个副本
// 对于每个元素，标签集合取并集，但排除对方 tombstones 中的标签
// 这样一侧的删除不会被另一侧的旧状态"复活"
func (s *ORSet) Merge(other *ORSet) {
    s.mu.Lock()
    defer s.mu.Unlock()

    // 合并 tombstones（两侧删除的并集）
    allTombstones := make(map[string]bool)
    for t := range s.tombstones {
        allTombstones[t] = true
    }
    for t := range other.tombstones {
        allTombstones[t] = true
    }

    // 合并 elements：取并集，但排除已在 tombstones 中的标签
    merged := make(map[string]map[string]bool)
    for elem, tags := range s.elements {
        for tag := range tags {
            if !allTombstones[tag] {
                if merged[elem] == nil {
                    merged[elem] = make(map[string]bool)
                }
                merged[elem][tag] = true
            }
        }
    }
    for elem, tags := range other.elements {
        for tag := range tags {
            if !allTombstones[tag] {
                if merged[elem] == nil {
                    merged[elem] = make(map[string]bool)
                }
                merged[elem][tag] = true
            }
        }
    }

    s.elements = merged
    s.tombstones = allTombstones
}

func main() {
    a := NewORSet()
    b := NewORSet()

    // 节点 A 添加 "x" 和 "y"
    a.Add("x") // tag-1
    a.Add("y") // tag-2

    // 把 A 的状态同步到 B
    b.Merge(a)

    // 并发操作：
    // 节点 B 删除 "x"（删除 tag-1）
    b.Remove("x")
    // 节点 A 再次添加 "x"（生成 tag-3，B 没观察到）
    a.Add("x") // tag-3

    fmt.Printf("合并前: A 包含 x? %v\n", a.Contains("x")) // true
    fmt.Printf("合并前: B 包含 x? %v\n", b.Contains("x")) // false

    // 合并：B 合入 A 的状态
    b.Merge(a)

    fmt.Printf("合并后: B 包含 x? %v\n", b.Contains("x")) // true —— tag-3 没被删除
    fmt.Printf("合并后: B 包含 y? %v\n", b.Contains("y")) // true
    fmt.Printf("合并后: B 所有元素: %v\n", b.Values())
}

// ShoppingCart 是一个 Map CRDT，key 为商品 ID，value 为 PN-Counter
type ShoppingCart struct {
    mu    sync.Mutex
    items map[string]*PNCounter // product_id → 数量
    nodeID int
    nodes  int
}

// AddItem 增加商品数量
func (c *ShoppingCart) AddItem(productID string, qty int) {
    c.mu.Lock()
    defer c.mu.Unlock()
    if c.items[productID] == nil {
        c.items[productID] = NewPNCounter(c.nodeID, c.nodes)
    }
    for i := 0; i < qty; i++ {
        c.items[productID].Inc()
    }
}

// Merge 合并另一端的购物车
func (c *ShoppingCart) Merge(other *ShoppingCart) {
    c.mu.Lock()
    defer c.mu.Unlock()
    for pid, counter := range other.items {
        if c.items[pid] == nil {
            c.items[pid] = NewPNCounter(c.nodeID, c.nodes)
        }
        c.items[pid].Merge(counter)
    }
}