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

feat: support direct network exit for runtimes

Browse Source
This commit is contained in:
rogee
2026-09-01 19:12:48 +08:00
parent 5674cfe8c3
commit 3729ce910f
15 changed files with 259 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/hub/environment.go
+7 -6
View File
@@ -355,7 +355,7 @@ func invalidateAccountsForExit(ctx context.Context, tx *sql.Tx, exitID string) (
func (s *Store) CreateBoundEnv(ctx context.Context, env Env, accountID, exitID string) (EnvironmentContext, bool, error) {
env.Alias, env.Name = strings.TrimSpace(env.Alias), strings.TrimSpace(env.Name)
if !aliasPattern.MatchString(env.Alias) || !validDisplayName(env.Name) || !aliasPattern.MatchString(accountID) ||
!exitIDPattern.MatchString(exitID) || !gatewayNamePattern.MatchString(env.Gateway) || !imageVersionPattern.MatchString(env.ImageVersion) ||
(exitID != "" && !exitIDPattern.MatchString(exitID)) || !gatewayNamePattern.MatchString(env.Gateway) || !imageVersionPattern.MatchString(env.ImageVersion) ||
env.Fingerprint.ProxyServer != "" {
return EnvironmentContext{}, false, ErrInvalid
}
@@ -391,15 +391,16 @@ func (s *Store) CreateBoundEnv(ctx context.Context, env Env, accountID, exitID s
if err := tx.QueryRowContext(ctx, `
INSERT INTO browser_env (alias, name, gateway_name, image_version, fingerprint)
SELECT $1, $2, $3, image.version, $5
FROM browser_image image, social_account account, network_exit network
FROM browser_image image, social_account account
WHERE image.version = $4 AND image.enabled AND account.id = $6 AND account.status = 'paused'
AND account.authorization_status = 'authorized' AND network.id = $7 AND network.health_status = 'healthy'
AND account.authorization_status = 'authorized'
AND ($7 = '' OR EXISTS (SELECT 1 FROM network_exit WHERE id = $7 AND health_status = 'healthy'))
RETURNING alias`, env.Alias, env.Name, env.Gateway, env.ImageVersion, encoded, accountID, exitID).Scan(&created); err != nil {
return EnvironmentContext{}, false, rowError(err)
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO environment_binding (id, account_id, browser_env_alias, network_exit_id)
VALUES ($1, $1, $2, $3)`, accountID, env.Alias, exitID); err != nil {
VALUES ($1, $1, $2, NULLIF($3, ''))`, accountID, env.Alias, exitID); err != nil {
return EnvironmentContext{}, false, publicDatabaseError(err)
}
if _, err := tx.ExecContext(ctx, `UPDATE social_account SET version = version + 1, updated_at = now() WHERE id = $1`, accountID); err != nil {
@@ -614,8 +615,8 @@ func (s *Store) ActivateRuntime(ctx context.Context, alias, runtimeID string, bi
if len(networkIDs) == 1 {
networkID = networkIDs[0]
}
if !aliasPattern.MatchString(alias) || !exitIDPattern.MatchString(runtimeID) || bindingVersion < 1 || !exitIDPattern.MatchString(exitID) ||
!exitIDPattern.MatchString(networkID) || len(networkIDs) != 1 {
if !aliasPattern.MatchString(alias) || !exitIDPattern.MatchString(runtimeID) || bindingVersion < 1 ||
(exitID != "" && !exitIDPattern.MatchString(exitID)) || !exitIDPattern.MatchString(networkID) || len(networkIDs) != 1 {
return EnvironmentContext{}, ErrInvalid
}
tx, err := s.db.BeginTx(ctx, nil)
internal/phasea/store.go
+24 -24
View File
@@ -858,15 +858,14 @@ func (s *Store) EnqueueConfirmation(ctx context.Context, confirmationID string)
}
return Task{}, false, publicDatabaseError(err)
}
if !networkExitID.Valid {
return Task{}, false, &ReadinessError{Reason: "network_exit_missing", Unavailable: true}
}
var exitStatus string
if err := tx.QueryRowContext(ctx, `SELECT health_status FROM network_exit WHERE id = $1 FOR SHARE`, networkExitID.String).Scan(&exitStatus); err != nil {
return Task{}, false, publicDatabaseError(err)
}
if exitStatus != "healthy" {
return Task{}, false, &ReadinessError{Reason: "network_exit_unhealthy", Unavailable: true}
if networkExitID.Valid {
var exitStatus string
if err := tx.QueryRowContext(ctx, `SELECT health_status FROM network_exit WHERE id = $1 FOR SHARE`, networkExitID.String).Scan(&exitStatus); err != nil {
return Task{}, false, publicDatabaseError(err)
}
if exitStatus != "healthy" {
return Task{}, false, &ReadinessError{Reason: "network_exit_unhealthy", Unavailable: true}
}
}
if cleanupPending {
return Task{}, false, &ReadinessError{Reason: "runtime_stop_pending", Unavailable: true}
@@ -1035,8 +1034,8 @@ func (s *Store) ResumeAccount(ctx context.Context, accountID string) error {
if err := tx.QueryRowContext(ctx, `
SELECT EXISTS (
SELECT 1 FROM environment_binding binding
JOIN network_exit network ON network.id = binding.network_exit_id
WHERE binding.account_id = $1 AND network.health_status = 'healthy'
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
WHERE binding.account_id = $1 AND (binding.network_exit_id IS NULL OR network.health_status = 'healthy')
AND NOT binding.runtime_cleanup_pending
AND NOT EXISTS (
SELECT 1 FROM runtime_instance runtime
@@ -1184,7 +1183,7 @@ func (s *Store) ResumeTask(ctx context.Context, taskID string) error {
JOIN confirmation confirmation ON confirmation.id = task.confirmation_id
JOIN environment_binding binding ON binding.account_id = task.account_id
JOIN browser_env environment ON environment.alias = binding.browser_env_alias
JOIN network_exit network ON network.id = binding.network_exit_id
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
WHERE task.id = $1 AND task.state IN ('policy_hold', 'needs_confirmation')
AND task.verification_result = 'not_executed'
@@ -1194,9 +1193,9 @@ func (s *Store) ResumeTask(ctx context.Context, taskID string) error {
AND confirmation.account_id = task.account_id AND confirmation.account_version = task.account_version
AND confirmation.draft_id = task.draft_id AND confirmation.draft_version = task.draft_version
AND confirmation.version = task.confirmation_version
AND network.health_status = 'healthy' AND NOT binding.runtime_cleanup_pending
AND (binding.network_exit_id IS NULL OR network.health_status = 'healthy') AND NOT binding.runtime_cleanup_pending
AND runtime.binding_version = binding.version AND runtime.lease_until > now()
FOR UPDATE OF task, account, draft, confirmation, binding, environment, network, runtime`, taskID).
FOR UPDATE OF task, account, draft, confirmation, binding, environment, runtime`, taskID).
Scan(&accountID, &confirmationID, &confirmationVersion)
if errors.Is(err, sql.ErrNoRows) {
task, taskErr := scanTask(tx.QueryRowContext(ctx, `
@@ -1255,8 +1254,8 @@ func taskReadinessReason(ctx context.Context, queryer rowQuerier, taskID string)
OR confirmation.version <> task.confirmation_version THEN 'confirmation_version_changed'
WHEN binding.id IS NULL THEN 'binding_missing'
WHEN environment.alias IS NULL THEN 'environment_missing'
WHEN network.id IS NULL THEN 'exit_missing'
WHEN network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
WHEN runtime.id IS NULL THEN 'runtime_missing'
WHEN runtime.binding_version IS DISTINCT FROM binding.version THEN 'binding_version_changed'
@@ -1382,7 +1381,7 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
JOIN confirmation c ON c.id = t.confirmation_id
JOIN environment_binding binding ON binding.account_id = a.id
JOIN browser_env environment ON environment.alias = binding.browser_env_alias
JOIN network_exit network ON network.id = binding.network_exit_id
LEFT JOIN network_exit network ON network.id = binding.network_exit_id
JOIN runtime_instance runtime ON runtime.binding_id = binding.id
AND runtime.released_at IS NULL AND runtime.lease_until > now() AND runtime.binding_version = binding.version
WHERE t.state = 'queued' AND a.status = 'active' AND a.authorization_status = 'authorized'
@@ -1391,9 +1390,9 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
AND c.account_id = t.account_id AND c.account_version = t.account_version
AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version
AND c.version = t.confirmation_version
AND network.health_status = 'healthy' AND NOT binding.runtime_cleanup_pending
AND (binding.network_exit_id IS NULL OR network.health_status = 'healthy') AND NOT binding.runtime_cleanup_pending
ORDER BY t.created_at, t.id
FOR UPDATE OF t, a, binding, network, runtime SKIP LOCKED LIMIT 1
FOR UPDATE OF t, a, binding, runtime SKIP LOCKED LIMIT 1
)
UPDATE operation_task t SET state = 'executing', hold_reason = NULL,
verification_result = NULL, verified_at = NULL, verified_by = NULL,
@@ -1448,8 +1447,8 @@ func (s *Store) complete(ctx context.Context, workerID string, execution Executi
OR confirmation.draft_version <> task.draft_version OR confirmation.version <> task.confirmation_version THEN 'confirmation_version_changed'
WHEN binding.id IS NULL THEN 'binding_missing'
WHEN environment.alias IS NULL THEN 'environment_missing'
WHEN network.id IS NULL THEN 'exit_missing'
WHEN network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
WHEN runtime.id IS NULL OR runtime.lease_until <= now() THEN 'runtime_missing'
WHEN claim.binding_version IS DISTINCT FROM binding.version
@@ -1600,8 +1599,8 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition,
OR confirmation.version <> t.confirmation_version THEN 'confirmation_version_changed'
WHEN binding.id IS NULL THEN 'binding_missing'
WHEN environment.alias IS NULL THEN 'environment_missing'
WHEN network.id IS NULL THEN 'exit_missing'
WHEN network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.network_exit_id IS NOT NULL AND network.id IS NULL THEN 'exit_missing'
WHEN binding.network_exit_id IS NOT NULL AND network.health_status <> 'healthy' THEN 'exit_unhealthy'
WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
WHEN runtime.id IS NULL THEN 'runtime_missing'
WHEN runtime.binding_version IS DISTINCT FROM binding.version THEN 'binding_version_changed'
@@ -1622,7 +1621,8 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition,
OR confirmation.id IS NULL OR confirmation.account_id <> t.account_id
OR confirmation.account_version <> t.account_version OR confirmation.draft_id <> t.draft_id
OR confirmation.draft_version <> t.draft_version OR confirmation.version <> t.confirmation_version
OR binding.id IS NULL OR environment.alias IS NULL OR network.id IS NULL OR network.health_status <> 'healthy'
OR binding.id IS NULL OR environment.alias IS NULL
OR (binding.network_exit_id IS NOT NULL AND (network.id IS NULL OR network.health_status <> 'healthy'))
OR binding.runtime_cleanup_pending OR runtime.id IS NULL
OR runtime.binding_version IS DISTINCT FROM binding.version OR runtime.lease_until <= now()
)