feat: support previous schema hash for smooth rolling upgrades

Add previousHash tracking to accept events from not-yet-upgraded nodes
during rolling upgrades when multiple nodes have publish=true.

Changes:
- SchemaCache tracks previousHash when schema changes
- Schema validation accepts events matching current OR previous hash
- SchemaRegistry publishes both hashes for cluster visibility
- Update documentation to reflect auto-resume and previousHash behavior

Author: wongfei2009

db/schema_cache.go

@@ -10,6 +10,7 @@ import (
 type SchemaCache struct {
 	mu            sync.RWMutex
 	schemaHash    string
+	previousHash  string // Hash before last schema change (for rolling upgrades)
 	schemaManager *SchemaManager
 	tables        []string
 }
@@ -41,8 +42,17 @@ func (sc *SchemaCache) GetSchemaHash() string {
 	return sc.schemaHash
 }
 
+// GetPreviousHash returns the previous schema hash (O(1))
+// Used during rolling upgrades to accept events from nodes not yet upgraded
+func (sc *SchemaCache) GetPreviousHash() string {
+	sc.mu.RLock()
+	defer sc.mu.RUnlock()
+	return sc.previousHash
+}
+
 // Recompute recalculates the schema hash from the database
 // Called during pause state to detect if local DDL has been applied
+// When schema changes, the old hash is preserved as previousHash
 func (sc *SchemaCache) Recompute(ctx context.Context) (string, error) {
 	sc.mu.Lock()
 	defer sc.mu.Unlock()
@@ -51,6 +61,11 @@ func (sc *SchemaCache) Recompute(ctx context.Context) (string, error) {
 	if err != nil {
 		return "", fmt.Errorf("recomputing schema hash: %w", err)
 	}
+
+	// Preserve old hash as previous when schema changes
+	if hash != sc.schemaHash && sc.schemaHash != "" {
+		sc.previousHash = sc.schemaHash
+	}
 	sc.schemaHash = hash
 	return hash, nil
 }

db/schema_cache_test.go

@@ -0,0 +1,136 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSchemaCache_PreviousHash(t *testing.T) {
+	// Create a test database
+	db, err := sql.Open("sqlite3", ":memory:")
+	require.NoError(t, err)
+	defer db.Close()
+
+	// Create initial schema
+	_, err = db.Exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)`)
+	require.NoError(t, err)
+
+	// Create schema manager
+	sm, err := NewSchemaManager(db)
+	require.NoError(t, err)
+
+	// Initialize schema cache
+	sc := NewSchemaCache()
+	ctx := context.Background()
+	err = sc.Initialize(ctx, sm, []string{"users"})
+	require.NoError(t, err)
+
+	// Get initial hash
+	initialHash := sc.GetSchemaHash()
+	assert.NotEmpty(t, initialHash)
+
+	// Previous hash should be empty initially
+	assert.Empty(t, sc.GetPreviousHash())
+
+	// Modify the schema
+	_, err = db.Exec(`ALTER TABLE users ADD COLUMN email TEXT`)
+	require.NoError(t, err)
+
+	// Recompute the hash
+	newHash, err := sc.Recompute(ctx)
+	require.NoError(t, err)
+
+	// New hash should be different
+	assert.NotEqual(t, initialHash, newHash)
+
+	// Previous hash should now be the initial hash
+	assert.Equal(t, initialHash, sc.GetPreviousHash())
+
+	// Current hash should be the new hash
+	assert.Equal(t, newHash, sc.GetSchemaHash())
+}
+
+func TestSchemaCache_PreviousHashUnchangedWhenSameSchema(t *testing.T) {
+	// Create a test database
+	db, err := sql.Open("sqlite3", ":memory:")
+	require.NoError(t, err)
+	defer db.Close()
+
+	// Create initial schema
+	_, err = db.Exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)`)
+	require.NoError(t, err)
+
+	// Create schema manager
+	sm, err := NewSchemaManager(db)
+	require.NoError(t, err)
+
+	// Initialize schema cache
+	sc := NewSchemaCache()
+	ctx := context.Background()
+	err = sc.Initialize(ctx, sm, []string{"users"})
+	require.NoError(t, err)
+
+	initialHash := sc.GetSchemaHash()
+
+	// Recompute without schema change
+	newHash, err := sc.Recompute(ctx)
+	require.NoError(t, err)
+
+	// Hash should be the same
+	assert.Equal(t, initialHash, newHash)
+
+	// Previous hash should still be empty (no change occurred)
+	assert.Empty(t, sc.GetPreviousHash())
+}
+
+func TestSchemaCache_MultipleSchemaChanges(t *testing.T) {
+	// Create a test database
+	db, err := sql.Open("sqlite3", ":memory:")
+	require.NoError(t, err)
+	defer db.Close()
+
+	// Create initial schema
+	_, err = db.Exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)`)
+	require.NoError(t, err)
+
+	// Create schema manager
+	sm, err := NewSchemaManager(db)
+	require.NoError(t, err)
+
+	// Initialize schema cache
+	sc := NewSchemaCache()
+	ctx := context.Background()
+	err = sc.Initialize(ctx, sm, []string{"users"})
+	require.NoError(t, err)
+
+	hash1 := sc.GetSchemaHash()
+
+	// First schema change
+	_, err = db.Exec(`ALTER TABLE users ADD COLUMN email TEXT`)
+	require.NoError(t, err)
+
+	hash2, err := sc.Recompute(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, hash1, sc.GetPreviousHash())
+
+	// Second schema change
+	_, err = db.Exec(`ALTER TABLE users ADD COLUMN phone TEXT`)
+	require.NoError(t, err)
+
+	hash3, err := sc.Recompute(ctx)
+	require.NoError(t, err)
+
+	// Previous hash should now be hash2 (only tracks one version back)
+	assert.Equal(t, hash2, sc.GetPreviousHash())
+	assert.Equal(t, hash3, sc.GetSchemaHash())
+
+	// Verify all three hashes are different
+	assert.NotEqual(t, hash1, hash2)
+	assert.NotEqual(t, hash2, hash3)
+	assert.NotEqual(t, hash1, hash3)
+}

db/sqlite.go

@@ -386,6 +386,14 @@ func (conn *SqliteStreamDB) GetSchemaHash() string {
 	return conn.schemaCache.GetSchemaHash()
 }
 
+// GetPreviousHash returns the previous schema hash (for rolling upgrades)
+func (conn *SqliteStreamDB) GetPreviousHash() string {
+	if conn.schemaCache == nil {
+		return ""
+	}
+	return conn.schemaCache.GetPreviousHash()
+}
+
 // GetSchemaCache returns the schema cache for direct access
 func (conn *SqliteStreamDB) GetSchemaCache() *SchemaCache {
 	return conn.schemaCache

docs/docs/architecture.md

@@ -373,11 +373,14 @@ HarmonyLite maintains a NATS KeyValue registry (`harmonylite-schema-registry`) t
 {
   "node_id": 1,
   "schema_hash": "a1b2c3d4e5f6...",
+  "previous_hash": "x9y8z7w6v5u4...",
   "harmonylite_version": "1.2.0",
   "updated_at": "2025-01-20T10:30:00Z"
 }
 ```
 
+The `previous_hash` field enables smooth rolling upgrades: a node accepts events matching either its current hash or its previous hash. This prevents unnecessary pauses when upgrading nodes one at a time in a multi-publisher cluster.
+
 This enables operators to monitor schema rollout progress across the cluster.
 
 ### Rolling Upgrade Flow
@@ -389,21 +392,30 @@ sequenceDiagram
     participant N2 as Node 2 (Old Schema)
     participant N3 as Node 3 (Old Schema)
     
-    Note over N1: Schema upgraded, new hash computed
-    N1->>NATS: Publish change (hash: abc123)
+    Note over N1: Schema upgraded, new hash H2 computed
+    Note over N1: Previous hash H1 preserved
+    N1->>NATS: Publish change (hash: H2)
     
-    NATS->>N2: Push message (hash: abc123)
-    Note over N2: Local hash: xyz789 ≠ abc123
+    NATS->>N2: Push message (hash: H2)
+    Note over N2: Local hash: H1 ≠ H2, no previous hash
     N2->>NATS: NAK with delay (pause)
     
-    Note over N2: Operator upgrades schema
-    Note over N2: Restart HarmonyLite
+    Note over N2: Operator applies DDL (no restart needed)
+    Note over N2: HarmonyLite auto-detects within 5 min
+    Note over N2: New hash: H2, previous: H1
     
-    NATS->>N2: Redeliver message (hash: abc123)
-    Note over N2: Local hash: abc123 = abc123
+    NATS->>N2: Redeliver message (hash: H2)
+    Note over N2: Local hash: H2 = H2
     N2->>N2: Apply change
     N2->>NATS: ACK
     
+    Note over N1,N2: Meanwhile, N2 still publishes with H1
+    N2->>NATS: Publish change (hash: H1)
+    NATS->>N1: Push message (hash: H1)
+    Note over N1: H1 = previous_hash, accept!
+    N1->>N1: Apply change
+    N1->>NATS: ACK
+    
     Note over N3: Same process repeats
 ```