LinkPort/internal/gateway/live_update.go

package gateway

import (
	"errors"
	"linkport/internal/model"
	"net"
	"reflect"
)

// ErrConfigChanged reports a failed live-configuration compare-and-swap.
var ErrConfigChanged = errors.New("gateway configuration changed; reload and retry")

// ConfigUpdate carries the snapshot used to construct Next. The From methods
// compare Current with the live snapshot before performing any persistence.
type ConfigUpdate struct {
	Current model.Proxy
	Next    model.Proxy
}

type preparedConfigUpdate struct {
	instance    *instance
	config      model.Proxy
	handler     model.Handler
	replacement net.Listener
}

// StartFrom loads and persists the latest enabled configuration while holding
// the lifecycle lock. The listener is published only after persistence succeeds.
func (m *Manager) StartFrom(load func() (model.Proxy, error), persist func(model.Proxy) error) (model.Proxy, error) {
	if load == nil || persist == nil {
		return model.Proxy{}, errors.New("load and persist callbacks are required")
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	p, err := load()
	if err != nil {
		return p, err
	}
	p.Enabled = true
	prepared, err := m.prepareStartLocked(p)
	if err != nil {
		return p, err
	}
	if err = persist(p); err != nil {
		prepared.listener.Close()
		if prepared.managementListener != nil {
			prepared.managementListener.Close()
		}
		return p, err
	}
	m.publishStartLocked(prepared)
	return p, nil
}

// StopAndPersist persists the latest disabled snapshot before stopping. A
// persistence failure leaves a running gateway untouched.
func (m *Manager) StopAndPersist(id string, load func() (model.Proxy, error), persist func(model.Proxy) error) (model.Proxy, error) {
	if persist == nil {
		return model.Proxy{}, errors.New("persist callback is required")
	}
	m.mu.Lock()
	inst := m.instances[id]
	var p model.Proxy
	var err error
	if inst == nil {
		if load == nil {
			m.mu.Unlock()
			return model.Proxy{}, errors.New("load callback is required for a stopped gateway")
		}
		p, err = load()
		if err != nil {
			m.mu.Unlock()
			return p, err
		}
	} else {
		if inst.stopping {
			m.mu.Unlock()
			return model.Proxy{}, errors.New("gateway is already stopping")
		}
		p = inst.config
	}
	if p.ID != id {
		m.mu.Unlock()
		return p, errors.New("loaded gateway ID does not match")
	}
	p.Enabled = false
	if err = persist(p); err != nil {
		m.mu.Unlock()
		return p, err
	}
	m.snapshots[id] = p
	if inst == nil {
		delete(m.failures, id)
		m.mu.Unlock()
		return p, nil
	}
	inst.config = p
	inst.stopLocked()
	m.mu.Unlock()

	inst.waitStopped()
	m.mu.Lock()
	if m.instances[id] == inst {
		delete(m.instances, id)
	}
	delete(m.failures, id)
	m.mu.Unlock()
	return p, nil
}

// UpdateConfig persists and publishes one gateway configuration. A stopped
// gateway is only persisted; a running gateway is updated without revoking
// connections that already captured the previous configuration snapshot.
func (m *Manager) UpdateConfig(next model.Proxy, persist func() error) error {
	return m.updateConfigs([]ConfigUpdate{{Next: next}}, false, persist)
}

// UpdateConfigs atomically publishes resolved configuration snapshots after a
// single persistence commit. Listener changes are bound before persistence, so
// a bind or persistence failure leaves every running gateway unchanged.
func (m *Manager) UpdateConfigs(next []model.Proxy, persist func() error) error {
	updates := make([]ConfigUpdate, len(next))
	for idx := range next {
		updates[idx].Next = next[idx]
	}
	return m.updateConfigs(updates, false, persist)
}

// UpdateConfigFrom applies a compare-and-swap update for one gateway.
func (m *Manager) UpdateConfigFrom(current, next model.Proxy, persist func() error) error {
	return m.UpdateConfigsFrom([]ConfigUpdate{{Current: current, Next: next}}, persist)
}

// UpdateConfigsFrom atomically rejects the entire batch when any running
// gateway changed after its candidate configuration was constructed.
func (m *Manager) UpdateConfigsFrom(updates []ConfigUpdate, persist func() error) error {
	return m.updateConfigs(updates, true, persist)
}

func comparableConfig(config model.Proxy) model.Proxy {
	// Enabled controls restart persistence rather than a live connection. A
	// manual Start publishes the pre-commit false value, so it cannot be a CAS
	// token. Hooks are attached to per-connection copies at accept time.
	config.Enabled = false
	config.TargetDialer = nil
	config.PreparedTarget = nil
	// OutboundID is a persistence indirection. Saving an inline non-direct
	// outbound creates an ID, while the resolved transport remains identical.
	config.OutboundID = ""
	if config.Targets != nil {
		targets := make(map[string]model.Target, len(config.Targets))
		for id, target := range config.Targets {
			target.OutboundID = ""
			targets[id] = target
		}
		config.Targets = targets
	}
	return config
}

func (m *Manager) updateConfigs(changes []ConfigUpdate, compareCurrent bool, persist func() error) error {
	if persist == nil {
		return errors.New("persist callback is required")
	}
	m.mu.Lock()
	defer m.mu.Unlock()

	seen := make(map[string]struct{}, len(changes))
	updates := make([]preparedConfigUpdate, 0, len(changes))
	closeReplacements := func() {
		for idx := range updates {
			if updates[idx].replacement != nil {
				updates[idx].replacement.Close()
				updates[idx].replacement = nil
			}
		}
	}

	for _, change := range changes {
		config := change.Next
		if config.ID == "" {
			return errors.New("gateway ID is required")
		}
		if compareCurrent && change.Current.ID != config.ID {
			return errors.New("current and next gateway IDs must match")
		}
		if _, ok := seen[config.ID]; ok {
			return errors.New("duplicate gateway configuration")
		}
		seen[config.ID] = struct{}{}
		inst := m.instances[config.ID]
		if compareCurrent {
			observed, ok := m.snapshots[config.ID]
			if inst != nil {
				observed, ok = inst.config, true
			}
			if ok {
				if inst == nil {
					// Full configuration replacement is only allowed while every
					// gateway is stopped, so the store may legitimately be newer
					// than this lifecycle snapshot. Enabled still distinguishes a
					// stale update that was queued before StopAndPersist.
					if observed.Enabled != change.Current.Enabled {
						return ErrConfigChanged
					}
				} else if !reflect.DeepEqual(comparableConfig(observed), comparableConfig(change.Current)) {
					return ErrConfigChanged
				}
			}
		}
	}

	for _, change := range changes {
		config := change.Next
		inst := m.instances[config.ID]
		if inst == nil {
			continue
		}
		if inst.stopping {
			closeReplacements()
			return errors.New("gateway is stopping; retry after it stops")
		}
		if config.Type != inst.config.Type {
			closeReplacements()
			return errors.New("gateway protocol cannot change while running")
		}
		if config.ManagementListenPort != inst.config.ManagementListenPort || (inst.managementListener != nil && config.ListenHost != inst.config.ListenHost) {
			closeReplacements()
			return errors.New("stop the gateway before changing RabbitMQ management listen settings")
		}
		if config.ManagementListenPort != 0 && config.ManagementListenPort == config.ListenPort {
			closeReplacements()
			return errors.New("RabbitMQ management and AMQP listen ports must differ")
		}
		handler := m.handlers[config.Type]
		if handler == nil {
			closeReplacements()
			return errors.New("unsupported protocol")
		}
		if !config.HasDestination() {
			closeReplacements()
			return errors.New("select a target before updating the running gateway")
		}
		if err := config.ValidateTargetSelection(); err != nil {
			closeReplacements()
			return err
		}
		if err := config.ValidateGatewayHostKeys(); err != nil {
			closeReplacements()
			return err
		}
		address, err := listenAddress(config.ListenHost, config.ListenPort)
		if err != nil {
			closeReplacements()
			return err
		}
		update := preparedConfigUpdate{instance: inst, config: config, handler: handler}
		if config.ListenHost != inst.config.ListenHost || config.ListenPort != inst.config.ListenPort {
			update.replacement, err = net.Listen("tcp", address)
			if err != nil {
				closeReplacements()
				return listenStartError(address, err)
			}
		}
		updates = append(updates, update)
	}

	if err := persist(); err != nil {
		closeReplacements()
		return err
	}
	for _, change := range changes {
		if m.instances[change.Next.ID] == nil {
			m.snapshots[change.Next.ID] = change.Next
		}
	}
	for idx := range updates {
		update := &updates[idx]
		inst := update.instance
		published := update.config
		published.Enabled = inst.config.Enabled
		inst.config = published
		m.snapshots[published.ID] = published
		inst.persistent.Reconcile(published)
		delete(m.failures, update.config.ID)
		if update.replacement == nil {
			continue
		}
		old := inst.listener
		inst.listener = update.replacement
		inst.wg.Add(1)
		go m.accept(inst.ctx, inst, update.replacement, update.handler)
		update.replacement = nil
		old.Close()
	}
	return nil
}