go-sip/internal/rpc/execution_journal.go

package rpc

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"

	agentv1 "git.ipao.vip/rogee/go-sip/gen/agent/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
)

type savedOperation struct {
	Digest  string                            `json:"digest"`
	Receipt *agentv1.OperationReceipt         `json:"receipt"`
	Control *agentv1.ApplyTaskControlResponse `json:"control,omitempty"`
}
type savedExecution struct {
	Binding       *agentv1.ExecutionBinding `json:"binding"`
	Digest        string                    `json:"digest"`
	State         agentv1.ExecutionState    `json:"state"`
	Revision      int64                     `json:"revision"`
	CallState     string                    `json:"call_state"`
	ControlAction agentv1.ControlAction     `json:"control_action"`
}
type executionJournal struct {
	Version    int                       `json:"version"`
	Mode       string                    `json:"mode"`
	AgentID    string                    `json:"agent_id"`
	CellID     string                    `json:"cell_id"`
	Operations map[string]savedOperation `json:"operations"`
	Executions map[string]savedExecution `json:"executions"`
}

func (s *Server) loadExecutionJournal() error {
	if s.executionPath == "" {
		return nil
	}
	raw, err := os.ReadFile(s.executionPath)
	if errors.Is(err, os.ErrNotExist) {
		if _, sessionErr := os.Stat(s.sessions.statePath); sessionErr == nil {
			return errors.New("execution journal missing beside existing session journal")
		} else if !errors.Is(sessionErr, os.ErrNotExist) {
			return sessionErr
		}
		// Establish the empty execution journal before any activation can be
		// acknowledged; later absence must not silently erase execution history.
		return s.persistExecutionJournalLocked()
	}
	if err != nil {
		return err
	}
	decoder := json.NewDecoder(bytes.NewReader(raw))
	decoder.DisallowUnknownFields()
	var journal executionJournal
	if err := decoder.Decode(&journal); err != nil {
		return err
	}
	if err := decoder.Decode(new(any)); err != io.EOF {
		return errors.New("execution journal has trailing data")
	}
	if journal.Version != 1 || journal.Mode != s.mode || journal.AgentID != s.status.AgentId || journal.CellID != s.status.CellId || journal.Operations == nil || journal.Executions == nil {
		return errors.New("execution journal version or identity is invalid")
	}
	for key, record := range journal.Operations {
		if record.Receipt == nil || record.Receipt.Meta == nil || len(record.Digest) != 64 {
			return errors.New("invalid persisted operation")
		}
		if record.Control != nil && !proto.Equal(record.Control.Receipt, record.Receipt) {
			return errors.New("control receipt differs from operation receipt")
		}
		s.operations[key] = operationRecord{digest: record.Digest, receipt: record.Receipt, control: record.Control}
	}
	for id, record := range journal.Executions {
		if record.Binding == nil || record.Binding.ExecutionId != id || record.Revision != record.Binding.TaskRevision {
			return errors.New("invalid persisted execution binding")
		}
		if _, ok := agentv1.ExecutionState_name[int32(record.State)]; !ok {
			return errors.New("invalid persisted execution state")
		}
		if _, ok := agentv1.ControlAction_name[int32(record.ControlAction)]; !ok {
			return errors.New("invalid persisted control action")
		}
		execution := &executionRecord{binding: record.Binding, executeDigest: record.Digest, taskRevision: record.Revision, callState: record.CallState, controlAction: record.ControlAction, state: record.State}
		// A new process cannot infer Asterisk's state from an old local snapshot.
		// Never restore permits or clear unknown occupancy because a process restarted.
		if execution.state != agentv1.ExecutionState_EXECUTION_STATE_TERMINAL {
			execution.state = agentv1.ExecutionState_EXECUTION_STATE_UNKNOWN
			execution.unknown = true
		}
		s.executions[id] = execution
	}
	return nil
}

// Caller holds s.mu. Failure poisons further admission: in-memory state must
// never be acknowledged as durable after an unsuccessful journal write.
func (s *Server) persistExecutionJournalLocked() error {
	if s.executionErr != nil {
		return status.Errorf(codes.Internal, "execution journal unavailable: %v", s.executionErr)
	}
	if s.executionPath == "" {
		return nil
	}
	journal := executionJournal{Version: 1, Mode: s.mode, AgentID: s.status.AgentId, CellID: s.status.CellId, Operations: make(map[string]savedOperation, len(s.operations)), Executions: make(map[string]savedExecution, len(s.executions))}
	for key, record := range s.operations {
		journal.Operations[key] = savedOperation{Digest: record.digest, Receipt: record.receipt, Control: record.control}
	}
	for id, record := range s.executions {
		journal.Executions[id] = savedExecution{Binding: record.binding, Digest: record.executeDigest, State: record.state, Revision: record.taskRevision, CallState: record.callState, ControlAction: record.controlAction}
	}
	if err := writeRPCJournal(s.executionPath, journal); err != nil {
		s.executionErr = err
		return status.Errorf(codes.Internal, "persist execution journal: %v", err)
	}
	return nil
}
func (s *Server) executionJournalReady() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.executionErr != nil {
		return status.Errorf(codes.Internal, "execution journal unavailable: %v", s.executionErr)
	}
	return nil
}

func writeRPCJournal(path string, value any) error {
	directory := filepath.Dir(path)
	if err := os.MkdirAll(directory, 0700); err != nil {
		return err
	}
	file, err := os.CreateTemp(directory, ".rpc-journal-*")
	if err != nil {
		return err
	}
	name := file.Name()
	defer os.Remove(name)
	if err := file.Chmod(0600); err != nil {
		_ = file.Close()
		return err
	}
	if err := json.NewEncoder(file).Encode(value); err != nil {
		_ = file.Close()
		return err
	}
	if err := file.Sync(); err != nil {
		_ = file.Close()
		return err
	}
	if err := file.Close(); err != nil {
		return err
	}
	if err := os.Rename(name, path); err != nil {
		return err
	}
	dir, err := os.Open(directory)
	if err != nil {
		return err
	}
	syncErr := dir.Sync()
	closeErr := dir.Close()
	if err := errors.Join(syncErr, closeErr); err != nil {
		return fmt.Errorf("sync journal directory: %w", err)
	}
	return nil
}