rogee/go-sip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Require Dispatcher identity in Agent activation

Browse Source
This commit is contained in:
rogee
2026-09-30 07:57:53 +08:00
parent 5ed5409994
commit 58f23660b8
8 changed files with 25 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/dispatcher/agent.go
+7 -3
View File
@@ -11,6 +11,7 @@ import (
agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
"git.ipao.vip/rogee/go-sip/internal/configread"
"git.ipao.vip/rogee/go-sip/internal/store"
"git.ipao.vip/rogee/go-sip/internal/tenant"
"github.com/google/uuid"
"google.golang.org/protobuf/proto"
)
@@ -149,7 +150,10 @@ func (c *AgentCoordinator) Probe(ctx context.Context, agentID, cellID string) (*
return response.Status, nil
}
func (c *AgentCoordinator) Activate(ctx context.Context, agentID, cellID, bootID, epoch string, generation uint64) (AgentSession, error) {
func (c *AgentCoordinator) Activate(ctx context.Context, dispatcherID, agentID, cellID, bootID, epoch string, generation uint64) (AgentSession, error) {
if tenant.ValidateDispatcherID(dispatcherID) != nil {
return AgentSession{}, errors.New("Agent activation requires a canonical Dispatcher UUID v4")
}
client, err := c.client(agentID)
if err != nil {
return AgentSession{}, err
@@ -163,11 +167,11 @@ func (c *AgentCoordinator) Activate(ctx context.Context, agentID, cellID, bootID
}
operationID := "activate:" + activationID.String()
meta := &agentpb.RequestMeta{ProtocolVersion: "agent.v1", RequestId: operationID + ":request", TraceId: operationID, OperationId: operationID, DispatcherEpoch: epoch, AgentId: agentID, CellId: cellID, BootId: bootID}
response, err := client.ActivateAgent(ctx, &agentpb.ActivateAgentRequest{Meta: meta, Binding: &agentpb.AgentBinding{AgentId: agentID, CellId: cellID, ExpectedBootId: bootID, DispatcherEpoch: epoch, SessionGeneration: generation}, ActivationOperationId: operationID})
response, err := client.ActivateAgent(ctx, &agentpb.ActivateAgentRequest{Meta: meta, Binding: &agentpb.AgentBinding{AgentId: agentID, CellId: cellID, ExpectedBootId: bootID, DispatcherId: dispatcherID, DispatcherEpoch: epoch, SessionGeneration: generation}, ActivationOperationId: operationID})
if err != nil {
return AgentSession{}, err
}
if response.Session == nil || response.State != agentpb.ActivationState_ACTIVATION_STATE_ACTIVE {
if response == nil || response.Session == nil || response.State != agentpb.ActivationState_ACTIVATION_STATE_ACTIVE {
return AgentSession{}, errors.New("Agent activation was not active")
}
session := AgentSession{AgentID: agentID, CellID: cellID, BootID: bootID, DispatcherEpoch: epoch, SessionGeneration: response.Session.SessionGeneration, ExpiresAtUnixMs: response.Session.ExpiresAtUnixMs}
internal/dispatcher/agent_test.go
+11 -5
View File
@@ -15,10 +15,13 @@ import (
"google.golang.org/grpc/test/bufconn"
)
const approvedTestDispatcherID = "c046b893-8628-4589-ae50-619d049248a6"
func TestAgentCoordinatorProbesBeforeActivation(t *testing.T) {
listener := bufconn.Listen(1024 * 1024)
server := rpcserver.NewServer(rpcserver.ServerOptions{
Now: func() time.Time { return time.Date(2026, 9, 18, 1, 0, 0, 0, time.UTC) },
ApprovedDispatcherID: approvedTestDispatcherID,
Now: func() time.Time { return time.Date(2026, 9, 18, 1, 0, 0, 0, time.UTC) },
Status: &agentpb.AgentStatus{
AgentId: "agent-1",
CellId: "cell-1",
@@ -47,15 +50,18 @@ func TestAgentCoordinatorProbesBeforeActivation(t *testing.T) {
if status.BootId != "boot-probed" || status.SessionActive {
t.Fatalf("unexpected pre-activation status: %+v", status)
}
first, err := coordinator.Activate(context.Background(), "agent-1", "cell-1", status.BootId, "epoch-1", 0)
if _, err := coordinator.Activate(context.Background(), "not-a-dispatcher", "agent-1", "cell-1", status.BootId, "epoch-1", 0); err == nil {
t.Fatal("unapproved Dispatcher identity was admitted")
}
first, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-1", "cell-1", status.BootId, "epoch-1", 0)
if err != nil {
t.Fatal(err)
}
second, err := coordinator.Activate(context.Background(), "agent-1", "cell-1", status.BootId, "epoch-2", 0)
second, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-1", "cell-1", status.BootId, "epoch-2", 0)
if err != nil {
t.Fatal(err)
}
third, err := coordinator.Activate(context.Background(), "agent-1", "cell-1", status.BootId, "epoch-2", 0)
third, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-1", "cell-1", status.BootId, "epoch-2", 0)
if err != nil {
t.Fatal(err)
}
@@ -81,7 +87,7 @@ func TestAgentCoordinatorActivatesExecutesAndControlsWithoutRetryingOriginate(t
if err := coordinator.Register("agent-1", agentpb.NewAgentControlServiceClient(conn)); err != nil {
t.Fatal(err)
}
if _, err := coordinator.Activate(context.Background(), "agent-1", "cell-1", "boot-1", "epoch-1", 1); err != nil {
if _, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-1", "cell-1", "boot-1", "epoch-1", 1); err != nil {
t.Fatal(err)
}
internal/dispatcher/dispatcher_test.go
+1 -1
View File
@@ -266,7 +266,7 @@ func TestExecuteReservedBindsQuotaAndAgentExecution(t *testing.T) {
if err := coordinator.Register("agent-1", client); err != nil {
t.Fatal(err)
}
if _, err := coordinator.Activate(context.Background(), "agent-1", "cell-1", "boot-1", "epoch-1", 1); err != nil {
if _, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-1", "cell-1", "boot-1", "epoch-1", 1); err != nil {
t.Fatal(err)
}
result, err := d.ExecuteReserved(context.Background(), coordinator, "agent-1", task, raw, "reservation-integration", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
internal/dispatcher/local_flow_test.go
+1 -1
View File
@@ -88,7 +88,7 @@ func TestLocalContractBackedFlowEvidence(t *testing.T) {
if err := coordinator.Register("agent-a", client); err != nil {
t.Fatal(err)
}
if _, err := coordinator.Activate(context.Background(), "agent-a", "cell-a", "boot-a", "epoch-a", 1); err != nil {
if _, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-a", "cell-a", "boot-a", "epoch-a", 1); err != nil {
t.Fatal(err)
}
result, err := d.ExecuteReserved(context.Background(), coordinator, "agent-a", task, callRaw, "reservation-local-flow", snapshot.Digest)
internal/dispatcher/local_origination_rpc_test.go
+1 -1
View File
@@ -41,7 +41,7 @@ func TestLocalAuthorizedOriginationReachesOnlyMockAgentAfterFinalGate(t *testing
if err := coordinator.Register("agent-1", agentpb.NewAgentControlServiceClient(connection)); err != nil {
t.Fatal(err)
}
if _, err := coordinator.Activate(context.Background(), "agent-1", "cell-1", "boot-1", "epoch-1", 1); err != nil {
if _, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-1", "cell-1", "boot-1", "epoch-1", 1); err != nil {
t.Fatal(err)
}
executionID := acceptedLocalExecutionID(t, d, st, at, "command-agent-bridge")
internal/dispatcher/two_cell_test.go
+2 -2
View File
@@ -26,10 +26,10 @@ func TestTwoMockCellsKeepAgentSessionsAndPermitsSeparate(t *testing.T) {
if err := coordinator.Register("agent-b", clientB); err != nil {
t.Fatal(err)
}
if _, err := coordinator.Activate(context.Background(), "agent-a", "cell-a", "boot-a", "epoch-1", 1); err != nil {
if _, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-a", "cell-a", "boot-a", "epoch-1", 1); err != nil {
t.Fatal(err)
}
if _, err := coordinator.Activate(context.Background(), "agent-b", "cell-b", "boot-b", "epoch-1", 1); err != nil {
if _, err := coordinator.Activate(context.Background(), approvedTestDispatcherID, "agent-b", "cell-b", "boot-b", "epoch-1", 1); err != nil {
t.Fatal(err)
}