admin343/browser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: separate run builders from generated workers

Browse Source
This commit is contained in:
npc0-hue
2026-07-28 11:05:38 +08:00
parent 2ae27bb746
commit e739dd9c79
10 changed files with 257 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/service/control_test.go
+67 -1
View File
@@ -172,7 +172,7 @@ func TestCoreServiceRunHelloRejectsStalePackageKeyAfterReset(t *testing.T) {
}
pkg := readGeneratedPackageConfig(t, svc, session, distribution.ArtifactID)
hello := validRunControlHello()
hello.RunEndpointID = instance.RunEndpointID
hello.RunEndpointID = distribution.RunEndpointID
hello.RegistrationToken = pkg.AuthKey
hello.ServerInstanceID = instance.ID
hello.PluginID = instance.PluginID
@@ -202,6 +202,72 @@ func TestCoreServiceRunHelloRejectsStalePackageKeyAfterReset(t *testing.T) {
}
}
func TestCoreServiceRunHelloRejectsGeneratedRunOnPromotedBuildEndpoint(t *testing.T) {
svc, session, instance := newDistributionTestFixture(t)
builderID := instance.RunEndpointID
instance.State = domain.ServerInstanceStateFailed
if err := svc.store.ServerInstances().Update(instance); err != nil {
t.Fatalf("mark legacy server failed: %v", err)
}
if _, err := svc.GenerateRunDistributionForSession(session, domain.RunDistributionGenerateRequest{ServerInstanceID: instance.ID, TargetOS: "windows", TargetArch: "amd64", IdempotencyKey: "promoted-hello-fence"}); err != nil {
t.Fatalf("generate promoted Run: %v", err)
}
migrated, err := svc.GetServerInstance(instance.ID)
if err != nil {
t.Fatalf("get migrated server: %v", err)
}
key, plainKey, err := svc.ensureActiveComponentKey(instance.ID, domain.DistributionComponentRun, "")
if err != nil {
t.Fatalf("get component key: %v", err)
}
hello := validRunControlHello()
hello.RunEndpointID = builderID
hello.RegistrationToken = plainKey
hello.ServerInstanceID = instance.ID
hello.PluginID = instance.PluginID
hello.ComponentKind = domain.DistributionComponentRun
hello.KeyGeneration = key.Generation
if _, err := svc.RegisterRunHello(hello); err == nil || !strings.Contains(err.Error(), "does not match") {
t.Fatalf("expected shared builder registration rejection, got %v", err)
}
hello.RunEndpointID = migrated.RunEndpointID
if result, err := svc.RegisterRunHello(hello); err != nil || !result.Accepted {
t.Fatalf("expected dedicated Run registration acceptance, result=%+v err=%v", result, err)
}
}
func TestCoreServiceComponentRunCannotClaimDistributionBuild(t *testing.T) {
svc, session, instance := newDistributionTestFixture(t)
distribution, err := svc.GenerateRunDistributionForSession(session, domain.RunDistributionGenerateRequest{ServerInstanceID: instance.ID, TargetOS: "windows", TargetArch: "amd64", IdempotencyKey: "component-build-claim"})
if err != nil {
t.Fatalf("generate legacy Run: %v", err)
}
packageConfig := readGeneratedPackageConfig(t, svc, session, distribution.ArtifactID)
hello := validRunControlHello()
hello.RunEndpointID = instance.RunEndpointID
hello.RegistrationToken = packageConfig.AuthKey
hello.ServerInstanceID = instance.ID
hello.PluginID = instance.PluginID
hello.ComponentKind = domain.DistributionComponentRun
hello.KeyGeneration = packageConfig.KeyGeneration
registered, err := svc.RegisterRunHello(hello)
if err != nil || !registered.Accepted {
t.Fatalf("register legacy package: result=%+v err=%v", registered, err)
}
storedSession, err := svc.store.RunControlSessions().Get(instance.RunEndpointID)
if err != nil || !storedSession.RequireSignedRequests {
t.Fatalf("expected component session to require signatures, session=%+v err=%v", storedSession, err)
}
if activeSession := svc.runSessions[instance.RunEndpointID]; !activeSession.RequireSignedRequests {
t.Fatalf("expected in-memory component session to require signatures, session=%+v", activeSession)
}
claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: instance.RunEndpointID, SessionToken: registered.SessionToken, Capabilities: []string{domain.JobCapabilityDistributionBuild}, Capacity: domain.RunCapacity{MaxJobs: 1}})
if err != nil || claim.HasJob {
t.Fatalf("component Run must not claim distribution builds: claim=%+v err=%v", claim, err)
}
}
func TestCoreServiceRequestsCapabilityRefreshOnFingerprintDrift(t *testing.T) {
svc := newTestCoreService()
hello, err := svc.RegisterRunHello(validRunControlHello())
platform/service/distributions.go
+24
View File
@@ -44,6 +44,9 @@ func (svc *CoreService) GenerateRunDistributionForSession(sessionID string, requ
if err := svc.requireCompleteRuntimeBindings(user.ID, instance.ID, "run.generate.denied"); err != nil {
return domain.RunDistribution{}, err
}
if err := svc.promoteLegacyRunBinding(&instance); err != nil {
return domain.RunDistribution{}, err
}
builderEndpointID := instance.RunEndpointID
if strings.TrimSpace(instance.DeploymentTargetID) != "" {
builderEndpointID = instance.DeploymentTargetID
@@ -125,6 +128,27 @@ func (svc *CoreService) GenerateRunDistributionForSession(sessionID string, requ
return domain.CopyRunDistribution(distribution), nil
}
// promoteLegacyRunBinding reserves a server-scoped endpoint for a generated
// Run before a legacy server first requests a distribution. Its existing
// endpoint remains the trusted build target; reusing it in the package would
// allow the generated Run to replace the builder registration.
func (svc *CoreService) promoteLegacyRunBinding(instance *domain.ServerInstance) error {
if instance == nil || strings.TrimSpace(instance.DeploymentTargetID) != "" || (instance.State != domain.ServerInstanceStateDraft && instance.State != domain.ServerInstanceStateFailed) {
return nil
}
builderEndpointID := strings.TrimSpace(instance.RunEndpointID)
if builderEndpointID == "" {
return validationError("legacy Run generation requires a build target endpoint")
}
instance.DeploymentTargetID = builderEndpointID
instance.RunEndpointID = dedicatedRunEndpointID(instance.ID)
instance.UpdatedAt = svc.now()
if err := validator.ValidateServerInstance(*instance); err != nil {
return err
}
return svc.store.ServerInstances().Update(*instance)
}
func (svc *CoreService) GenerateClientManagerDistributionForSession(sessionID string, request domain.ClientManagerBuildRequest) (domain.ClientManagerDistribution, error) {
request = domain.CopyClientManagerBuildRequest(request)
if strings.TrimSpace(request.IdempotencyKey) == "" {
platform/service/distributions_test.go
+25
View File
@@ -120,6 +120,31 @@ func TestCoreServiceBuildsDedicatedRunOnDeploymentTarget(t *testing.T) {
}
}
func TestCoreServicePromotesLegacyRunBindingBeforeDistributionBuild(t *testing.T) {
svc, session, instance := newDistributionTestFixture(t)
builderID := instance.RunEndpointID
instance.State = domain.ServerInstanceStateFailed
if err := svc.store.ServerInstances().Update(instance); err != nil {
t.Fatalf("mark legacy server failed: %v", err)
}
distribution, err := svc.GenerateRunDistributionForSession(session, domain.RunDistributionGenerateRequest{ServerInstanceID: instance.ID, TargetOS: "windows", TargetArch: "amd64", IdempotencyKey: "legacy-promote-build"})
if err != nil {
t.Fatalf("generate promoted legacy Run: %v", err)
}
migrated, err := svc.GetServerInstance(instance.ID)
if err != nil {
t.Fatalf("get migrated server: %v", err)
}
if migrated.DeploymentTargetID != builderID || migrated.RunEndpointID != "server-run-"+instance.ID {
t.Fatalf("expected legacy binding promotion, got %+v", migrated)
}
job, err := svc.GetJob(distribution.BuildJobID)
if err != nil || job.RunEndpointID != builderID || distribution.RunEndpointID != migrated.RunEndpointID {
t.Fatalf("expected build target %q and dedicated package endpoint %q, job=%+v distribution=%+v err=%v", builderID, migrated.RunEndpointID, job, distribution, err)
}
}
func TestCoreServiceRejectsDistributionBuildForStaleRunEndpoint(t *testing.T) {
svc, session, instance := newDistributionTestFixture(t)
svc.now = func() time.Time { return fixedTime.Add(capacityHeartbeatStaleAfter + time.Second) }
platform/service/job_channel.go
+16
View File
@@ -29,6 +29,9 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai
if err != nil {
return domain.RunJobClaimResult{}, err
}
if session.RequireSignedRequests {
claim.Capabilities = withoutCapability(claim.Capabilities, domain.JobCapabilityDistributionBuild)
}
stamp := svc.now()
svc.jobMu.Lock()
@@ -40,6 +43,9 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai
if claim.Capacity.MaxJobs > 0 && claim.Capacity.RunningJobs >= claim.Capacity.MaxJobs {
return emptyJobClaim(claim.RunEndpointID, stamp), nil
}
if session.RequireSignedRequests && len(claim.Capabilities) == 0 {
return emptyJobClaim(claim.RunEndpointID, stamp), nil
}
jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: claim.RunEndpointID})
if err != nil {
return domain.RunJobClaimResult{}, err
@@ -82,6 +88,16 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai
}), nil
}
func withoutCapability(capabilities []string, forbidden string) []string {
filtered := make([]string, 0, len(capabilities))
for _, capability := range capabilities {
if capability != forbidden {
filtered = append(filtered, capability)
}
}
return filtered
}
func (svc *CoreService) AckRunJob(ack domain.RunJobAck) (domain.RunJobAckResult, error) {
if err := validator.ValidateRunJobAck(ack); err != nil {
return domain.RunJobAckResult{}, err