Merge pull request #39 from actionforge/p4

Add nodes for Perforce P4

Author: sebastianrath

.github/workflows/workflow.yml

@@ -101,7 +101,17 @@ jobs:
           runner-path: ${{ github.workspace }}/actrun
           graph-file: build-test-publish.act
           inputs: ${{ toJson(inputs) }}
-          secrets: ${{ toJson(secrets) }}
+          secrets: >-
+            {
+              "APPLE_P12_CERTIFICATE_PASSWORD": "${{ secrets.APPLE_P12_CERTIFICATE_PASSWORD }}",
+              "APPLE_P12_CERTIFICATE_BASE64": "${{ secrets.APPLE_P12_CERTIFICATE_BASE64 }}",
+              "TESTE2E_DO_S3_ACCESS_KEY": "${{ secrets.TESTE2E_DO_S3_ACCESS_KEY }}",
+              "TESTE2E_DO_S3_SECRET_KEY": "${{ secrets.TESTE2E_DO_S3_SECRET_KEY }}",
+              "TESTE2E_AWS_S3_ACCESS_KEY": "${{ secrets.TESTE2E_AWS_S3_ACCESS_KEY }}",
+              "TESTE2E_AWS_S3_SECRET_KEY": "${{ secrets.TESTE2E_AWS_S3_SECRET_KEY }}",
+              "PUBLISH_S3_SECRET_KEY": "${{ secrets.PUBLISH_S3_SECRET_KEY }}",
+              "PUBLISH_S3_ACCESS_KEY": "${{ secrets.PUBLISH_S3_ACCESS_KEY }}"
+            }
           matrix: ${{ toJson(matrix) }}
 
   docker-manifest:

agent/client.go

@@ -12,6 +12,7 @@ import (
 type Client struct {
 	serverURL  string
 	token      string
+	uuid       string
 	httpClient *http.Client
 }
 
@@ -25,6 +26,10 @@ func NewClient(serverURL, token string) *Client {
 	}
 }
 
+func (c *Client) SetUUID(uuid string) {
+	c.uuid = uuid
+}
+
 func (c *Client) doRequest(method, path string, body interface{}) (*http.Response, error) {
 	var bodyReader io.Reader
 	if body != nil {
@@ -40,6 +45,9 @@ func (c *Client) doRequest(method, path string, body interface{}) (*http.Respons
 		return nil, err
 	}
 	req.Header.Set("Authorization", "Bearer "+c.token)
+	if c.uuid != "" {
+		req.Header.Set("X-Agent-UUID", c.uuid)
+	}
 	if body != nil {
 		req.Header.Set("Content-Type", "application/json")
 	}

agent/flock_unix.go

@@ -0,0 +1,16 @@
+//go:build !windows
+
+package agent
+
+import (
+	"os"
+	"syscall"
+)
+
+func lockFileExclusive(f *os.File) error {
+	return syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+}
+
+func unlockFile(f *os.File) {
+	_ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
+}

agent/flock_windows.go

@@ -0,0 +1,45 @@
+//go:build windows
+
+package agent
+
+import (
+	"os"
+	"syscall"
+	"unsafe"
+)
+
+var (
+	modkernel32      = syscall.NewLazyDLL("kernel32.dll")
+	procLockFileEx   = modkernel32.NewProc("LockFileEx")
+	procUnlockFileEx = modkernel32.NewProc("UnlockFileEx")
+)
+
+const (
+	lockfileExclusiveLock   = 0x00000002
+	lockfileFailImmediately = 0x00000001
+)
+
+func lockFileExclusive(f *os.File) error {
+	var overlapped syscall.Overlapped
+	r1, _, err := procLockFileEx.Call(
+		uintptr(f.Fd()),
+		uintptr(lockfileExclusiveLock|lockfileFailImmediately),
+		0,
+		1, 0,
+		uintptr(unsafe.Pointer(&overlapped)),
+	)
+	if r1 == 0 {
+		return err
+	}
+	return nil
+}
+
+func unlockFile(f *os.File) {
+	var overlapped syscall.Overlapped
+	procUnlockFileEx.Call(
+		uintptr(f.Fd()),
+		0,
+		1, 0,
+		uintptr(unsafe.Pointer(&overlapped)),
+	)
+}

agent/vcs/p4.go

@@ -93,7 +93,7 @@ func (p *P4Provider) Checkout(ctx context.Context, url, ref, pipeline, destDir s
 			return CheckoutResult{}, fmt.Errorf("p4 sync pipeline file failed: %w", err)
 		}
 
-		return CheckoutResult{Dir: root, Persistent: true}, nil
+		return CheckoutResult{Dir: root, Persistent: true, P4Client: p.clientName}, nil
 	}
 
 	// Create temporary workspace
@@ -132,7 +132,7 @@ func (p *P4Provider) Checkout(ctx context.Context, url, ref, pipeline, destDir s
 		return CheckoutResult{}, fmt.Errorf("p4 sync pipeline file failed: %w", err)
 	}
 
-	return CheckoutResult{Dir: absDir}, nil
+	return CheckoutResult{Dir: absDir, Persistent: true, P4Client: p.clientName}, nil
 }
 
 func (p *P4Provider) Cleanup(ctx context.Context) error {

agent/vcs/vcs.go

@@ -33,6 +33,10 @@ type CheckoutResult struct {
 	Persistent bool
 	// SHA is the resolved commit SHA (or changelist number for P4) after checkout.
 	SHA string
+	// P4Client is the Perforce workspace name created or reused during checkout.
+	// The worker should set P4CLIENT in the subprocess environment so that
+	// p4 commands within the graph can operate on the same workspace.
+	P4Client string
 }
 
 // Provider handles VCS checkout operations.

agent/worker.go

@@ -26,59 +26,105 @@ type Worker struct {
 	vcsOpts      vcs.Options
 	pollInterval time.Duration
 	uuid         string
+	slotCleanup  func()
 	log          *logrus.Entry
 
 	metricsMu    sync.Mutex
 	lastCounters *RawCounters
 }
 
 func NewWorker(client *Client, docker DockerConfig, vcsOpts vcs.Options) *Worker {
+	uuid, cleanup := acquireAgentSlot()
+	client.SetUUID(uuid)
 	return &Worker{
 		client:       client,
 		docker:       docker,
 		vcsOpts:      vcsOpts,
 		pollInterval: 1 * time.Second,
-		uuid:         loadOrGenerateUUID(),
+		uuid:         uuid,
+		slotCleanup:  cleanup,
 		log:          logrus.WithField("component", "agent"),
 	}
 }
 
-// uuidFilePath returns the path to the persistent UUID file in the user's config directory.
-func uuidFilePath() string {
+// agentSlotDir returns the directory for agent UUID slot files.
+func agentSlotDir() string {
 	dir, err := os.UserConfigDir()
 	if err != nil {
 		dir = os.TempDir()
 	}
-	return filepath.Join(dir, "actionforge", "agent-uuid")
+	return filepath.Join(dir, "actionforge")
 }
 
-// loadOrGenerateUUID loads a persistent UUID from disk, or generates and saves a new one.
-func loadOrGenerateUUID() string {
-	path := uuidFilePath()
-	if data, err := os.ReadFile(path); err == nil {
-		if id := strings.TrimSpace(string(data)); len(id) == 36 {
-			return id
+// acquireAgentSlot finds and locks the lowest available agent slot.
+// Each slot has a persistent UUID file and a lock file. When the process
+// exits, the lock is released so the next process can reuse that slot
+// (and its UUID/metrics history).
+// Returns the UUID and a cleanup function that releases the lock.
+func acquireAgentSlot() (string, func()) {
+	dir := agentSlotDir()
+	_ = os.MkdirAll(dir, 0700)
+
+	const maxSlots = 256
+	for i := 0; i < maxSlots; i++ {
+		lockPath := filepath.Join(dir, fmt.Sprintf("agent-%d.lock", i))
+		uuidPath := filepath.Join(dir, fmt.Sprintf("agent-%d.uuid", i))
+
+		lockFile, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0600)
+		if err != nil {
+			continue
+		}
+
+		if err := lockFileExclusive(lockFile); err != nil {
+			if cerr := lockFile.Close(); cerr != nil {
+				logrus.WithError(cerr).Warn("failed to close lock file")
+			}
+			continue
 		}
+
+		// Slot acquired — read or generate UUID
+		uuid := ""
+		if data, err := os.ReadFile(uuidPath); err == nil {
+			if id := strings.TrimSpace(string(data)); len(id) == 36 {
+				uuid = id
+			}
+		}
+		if uuid == "" {
+			var buf [16]byte
+			_, _ = rand.Read(buf[:])
+			buf[6] = (buf[6] & 0x0f) | 0x40 // version 4
+			buf[8] = (buf[8] & 0x3f) | 0x80 // variant 1
+			uuid = fmt.Sprintf("%08x-%04x-%04x-%04x-%012x",
+				buf[0:4], buf[4:6], buf[6:8], buf[8:10], buf[10:16])
+			_ = os.WriteFile(uuidPath, []byte(uuid+"\n"), 0600)
+		}
+
+		cleanup := func() {
+			unlockFile(lockFile)
+			if cerr := lockFile.Close(); cerr != nil {
+				logrus.WithError(cerr).Warn("failed to close lock file")
+			}
+		}
+		return uuid, cleanup
 	}
 
-	// Generate UUID v4
+	// Fallback: all slots taken, generate ephemeral UUID with no lock
 	var buf [16]byte
 	_, _ = rand.Read(buf[:])
-	buf[6] = (buf[6] & 0x0f) | 0x40 // version 4
-	buf[8] = (buf[8] & 0x3f) | 0x80 // variant 1
-	id := fmt.Sprintf("%08x-%04x-%04x-%04x-%012x",
-		buf[0:4], buf[4:6], buf[6:8], buf[8:10], buf[10:16])
-
-	_ = os.MkdirAll(filepath.Dir(path), 0700)
-	_ = os.WriteFile(path, []byte(id+"\n"), 0600)
-	return id
+	buf[6] = (buf[6] & 0x0f) | 0x40
+	buf[8] = (buf[8] & 0x3f) | 0x80
+	return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x",
+		buf[0:4], buf[4:6], buf[6:8], buf[8:10], buf[10:16]), func() {}
 }
 
 // maxConsecutiveErrors is the number of consecutive connection errors before
 // Run returns ErrConnectionLost so the caller can decide to restart.
 const maxConsecutiveErrors = 10
 
 func (w *Worker) Run(ctx context.Context) error {
+	if w.slotCleanup != nil {
+		defer w.slotCleanup()
+	}
 	w.log.Info("starting")
 
 	// Take initial snapshot for delta computation
@@ -358,12 +404,16 @@ func (w *Worker) execute(ctx context.Context, job *ClaimResponse) {
 	env = append(env, "BUILD_TMPDIR="+tmpDir)
 	env = append(env, "BUILD_VCS_TYPE="+job.VCSType)
 	env = append(env, "BUILD_VCS_URL="+job.VCSURL)
+	env = append(env, "BUILD_REF="+ref)
 	if job.RepoID != "" {
 		env = append(env, "BUILD_REPO_ID="+job.RepoID)
 	}
 	if checkout.SHA != "" {
 		env = append(env, "BUILD_COMMIT_SHA="+checkout.SHA)
 	}
+	if checkout.P4Client != "" {
+		env = append(env, "P4CLIENT="+checkout.P4Client)
+	}
 
 	// Resolve env mappings from trigger config (if present)
 	if len(job.EnvMappings) > 0 && job.MatrixValues != nil {

core/base.go

@@ -49,6 +49,7 @@ const (
 	CredentialTypeSSH CredentialType = iota
 	CredentialTypeUsernamePassword
 	CredentialTypeAccessKey
+	CredentialTypeP4
 )
 
 type Credentials interface {