chore: merge develop into branch, resolve cmd/meridian/main.go conflict

Kept the full migration runner implementation from our branch over the
stub main() that was added to develop.

Author: bjcoombs

cmd/meridian/lock.go

@@ -0,0 +1,68 @@
+// Package main provides the unified Meridian binary entry point.
+package main
+
+import (
+	"context"
+	"sync"
+
+	"github.com/meridianhub/meridian/shared/platform/scheduler"
+)
+
+// Compile-time assertion: localLockManager satisfies the scheduler.DistributedLock interface.
+var _ scheduler.DistributedLock = (*localLockManager)(nil)
+
+// lockKey is a composite key for per-resource locks. Using a struct avoids
+// delimiter-based collisions when tenantID or resourceID contain ":".
+type lockKey struct {
+	tenantID   string
+	resourceID string
+}
+
+// localLockManager is an in-process mutex-based lock manager satisfying
+// the scheduler.DistributedLock interface. It is suitable for single-process
+// deployments where Redis is not available.
+type localLockManager struct {
+	mu        sync.Mutex
+	locks     map[lockKey]uint64
+	nextToken uint64
+}
+
+func newLocalLockManager() *localLockManager {
+	return &localLockManager{locks: make(map[lockKey]uint64)}
+}
+
+// Acquire attempts to acquire the lock for the given tenant and resource.
+// Returns (true, release, nil) if acquired, (false, nil, nil) if already held.
+// The release function is safe to call multiple times; only the first call from
+// the original acquirer has effect - stale releases are ignored via token check.
+func (m *localLockManager) Acquire(_ context.Context, tenantID, resourceID string) (bool, func(), error) {
+	k := lockKey{tenantID: tenantID, resourceID: resourceID}
+
+	m.mu.Lock()
+	if _, ok := m.locks[k]; ok {
+		m.mu.Unlock()
+		return false, nil, nil
+	}
+	m.nextToken++
+	token := m.nextToken
+	m.locks[k] = token
+	m.mu.Unlock()
+
+	release := func() {
+		m.mu.Lock()
+		if m.locks[k] == token {
+			delete(m.locks, k)
+		}
+		m.mu.Unlock()
+	}
+	return true, release, nil
+}
+
+// alwaysLeader is a no-op leader election stub that always reports the current
+// instance as leader. Valid for single-process deployments where there is only
+// one instance to coordinate.
+type alwaysLeader struct{}
+
+func (a *alwaysLeader) IsLeader() bool          { return true }
+func (a *alwaysLeader) Start(_ context.Context) {}
+func (a *alwaysLeader) Stop()                   {}

cmd/meridian/lock_test.go

@@ -0,0 +1,175 @@
+package main
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLocalLockManager_AcquireAndRelease(t *testing.T) {
+	m := newLocalLockManager()
+	ctx := context.Background()
+
+	acquired, release, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired)
+	assert.NotNil(t, release)
+
+	release()
+
+	// After release, can acquire again
+	acquired, release, err = m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired)
+	assert.NotNil(t, release)
+	release()
+}
+
+func TestLocalLockManager_DoubleAcquireReturnsFalse(t *testing.T) {
+	m := newLocalLockManager()
+	ctx := context.Background()
+
+	acquired1, release1, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired1)
+	defer release1()
+
+	// Second acquire on same key should fail
+	acquired2, release2, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.False(t, acquired2)
+	assert.Nil(t, release2)
+}
+
+func TestLocalLockManager_ReleaseOfUnheldLockIsIdempotent(t *testing.T) {
+	m := newLocalLockManager()
+	ctx := context.Background()
+
+	acquired, release, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired)
+
+	// Call release twice - should not panic
+	release()
+	release()
+
+	// Lock should be acquirable again
+	acquired2, release2, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired2)
+	release2()
+}
+
+func TestLocalLockManager_DifferentKeysAreIndependent(t *testing.T) {
+	m := newLocalLockManager()
+	ctx := context.Background()
+
+	acquired1, release1, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired1)
+	defer release1()
+
+	// Different resource on same tenant can be acquired
+	acquired2, release2, err := m.Acquire(ctx, "tenant-1", "resource-b")
+	require.NoError(t, err)
+	assert.True(t, acquired2)
+	defer release2()
+
+	// Different tenant, same resource can be acquired
+	acquired3, release3, err := m.Acquire(ctx, "tenant-2", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired3)
+	defer release3()
+}
+
+func TestLocalLockManager_StaleReleaseDoesNotUnlockNewHolder(t *testing.T) {
+	m := newLocalLockManager()
+	ctx := context.Background()
+
+	// First acquisition
+	acquired1, release1, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired1)
+
+	// Release - lock is now free
+	release1()
+
+	// Second acquisition of same key
+	acquired2, release2, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.True(t, acquired2)
+	defer release2()
+
+	// Stale release1 called again - must NOT release the new holder's lock
+	release1()
+
+	// Lock should still be held: a third acquire must fail
+	acquired3, release3, err := m.Acquire(ctx, "tenant-1", "resource-a")
+	require.NoError(t, err)
+	assert.False(t, acquired3, "stale release should not have freed the lock held by the second acquirer")
+	assert.Nil(t, release3)
+}
+
+func TestLocalLockManager_ColonInResourceIDNoCollision(t *testing.T) {
+	m := newLocalLockManager()
+	ctx := context.Background()
+
+	// These two pairs would produce the same string with a ":" delimiter:
+	// "tenant" + ":" + "a:b" == "tenant:a" + ":" + "b"
+	// The struct key prevents this collision.
+	acquired1, release1, err := m.Acquire(ctx, "tenant", "a:b")
+	require.NoError(t, err)
+	assert.True(t, acquired1)
+	defer release1()
+
+	acquired2, release2, err := m.Acquire(ctx, "tenant:a", "b")
+	require.NoError(t, err)
+	assert.True(t, acquired2, "keys with ':' in resourceID must not collide with different tenantID")
+	defer release2()
+}
+
+func TestAlwaysLeader_IsLeaderReturnsTrue(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	l := &alwaysLeader{}
+	l.Start(ctx)
+	assert.True(t, l.IsLeader())
+	l.Stop()
+	assert.True(t, l.IsLeader())
+}
+
+func TestLocalLockManager_Concurrency(t *testing.T) {
+	m := newLocalLockManager()
+	ctx := context.Background()
+
+	const goroutines = 50
+	acquisitions := make([]bool, goroutines)
+	var wg sync.WaitGroup
+
+	for i := range goroutines {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			acquired, release, err := m.Acquire(ctx, "tenant-1", "resource-a")
+			if err == nil && acquired {
+				acquisitions[idx] = true
+				release()
+			}
+		}(i)
+	}
+
+	wg.Wait()
+
+	// At least one goroutine should have acquired the lock
+	var count int
+	for _, a := range acquisitions {
+		if a {
+			count++
+		}
+	}
+	assert.Greater(t, count, 0)
+}