From 78b64a7a5d664a7c465b7ab6c7fef022b9a7c67e Mon Sep 17 00:00:00 2001 From: shrutiyam-glitch Date: Thu, 20 Aug 2026 15:17:08 -0700 Subject: [PATCH] Fix actoridentity flaky test --- .../internal/actoridentity/actoridentity.go | 79 +++++++++++++------ 1 file changed, 55 insertions(+), 24 deletions(-) diff --git a/cmd/ateapi/internal/actoridentity/actoridentity.go b/cmd/ateapi/internal/actoridentity/actoridentity.go index df1b9acbc..1beaefc36 100644 --- a/cmd/ateapi/internal/actoridentity/actoridentity.go +++ b/cmd/ateapi/internal/actoridentity/actoridentity.go @@ -317,61 +317,92 @@ func validateWorkerRef(worker *ateapipb.ObjectRef) error { // authorizeActor resolves the actor from the authenticated worker and verifies // that the worker and actor still point at one another. Actor identity supplied // by the requester never participates in this authorization decision. +// The worker is resolved from cache first (hot path), but denials fall back +// to the authoritative store to handle watch-delivery lag right after ResumeActor. func (s *Server) authorizeActor(ctx context.Context, caller *ateletCaller, req *ateapipb.MintCertRequest) (*ateapipb.Actor, resources.ActorRef, error) { - // Denials are deliberately indistinguishable from each other: a caller that - // is not entitled to a worker should not learn its assignment. - deny := func(reason string, args ...any) error { - slog.WarnContext(ctx, "ActorIdentity denied: "+reason, - append([]any{slog.String("worker", req.GetWorker().GetName()), slog.String("callerPod", caller.podName), slog.String("callerNode", caller.nodeName)}, args...)...) - return status.Errorf(codes.PermissionDenied, "caller is not permitted to mint credentials for this actor") - } - worker, err := s.workers.Worker(req.GetWorker().GetName()) if err != nil { if errors.Is(err, store.ErrNotFound) { - return nil, resources.ActorRef{}, deny("worker not found") + return nil, resources.ActorRef{}, s.denyMint(ctx, caller, req, "worker not found") } slog.ErrorContext(ctx, "ActorIdentity: failed to read worker", slog.Any("err", err)) return nil, resources.ActorRef{}, status.Error(codes.Internal, "failed to look up worker") } + + actor, actorRef, reason, err := s.authorizeWithWorker(ctx, worker, caller, req) + if err != nil { + return nil, resources.ActorRef{}, err // Internal error (e.g., actor lookup failed) + } + if reason == "" { + return actor, actorRef, nil // Success! + } + + // Read-through: re-check the authoritative worker from the store on denial. + fresh, ferr := s.store.GetWorker(ctx, req.GetWorker().GetName()) + if ferr != nil { + if !errors.Is(ferr, store.ErrNotFound) { + slog.ErrorContext(ctx, "ActorIdentity: read-through worker lookup failed", slog.Any("err", ferr)) + } + // The original denial stands + return nil, resources.ActorRef{}, s.denyMint(ctx, caller, req, reason) + } + + actor, actorRef, retryReason, retryErr := s.authorizeWithWorker(ctx, fresh, caller, req) + if retryErr != nil { + return nil, resources.ActorRef{}, retryErr + } + if retryReason != "" { + return nil, resources.ActorRef{}, s.denyMint(ctx, caller, req, retryReason) + } + + slog.InfoContext(ctx, "ActorIdentity: authorized via store read-through; worker cache was stale", + slog.String("worker", req.GetWorker().GetName())) + return actor, actorRef, nil +} + +// denyMint logs the internal reason and returns a uniform PermissionDenied. +func (s *Server) denyMint(ctx context.Context, caller *ateletCaller, req *ateapipb.MintCertRequest, reason string, args ...any) error { + slog.WarnContext(ctx, "ActorIdentity denied: "+reason, + append([]any{slog.String("worker", req.GetWorker().GetName()), slog.String("callerPod", caller.podName), slog.String("callerNode", caller.nodeName)}, args...)...) + return status.Errorf(codes.PermissionDenied, "caller is not permitted to mint credentials for this actor") +} + +// authorizeWithWorker returns a (reason string) instead of calling denyMint directly. +// If reason is non-empty, the authorization failed. +func (s *Server) authorizeWithWorker(ctx context.Context, worker *ateapipb.Worker, caller *ateletCaller, req *ateapipb.MintCertRequest) (*ateapipb.Actor, resources.ActorRef, string, error) { if worker.GetNodeName() != caller.nodeName { - return nil, resources.ActorRef{}, deny("worker is hosted on a different node", slog.String("workerNode", worker.GetNodeName())) + return nil, resources.ActorRef{}, "worker is hosted on a different node", nil } actorRef := resources.ActorRefFromObjectRef(worker.GetStatus().GetAssignment().GetActor()) if actorRef == (resources.ActorRef{}) { - return nil, resources.ActorRef{}, deny("worker has no actor assignment") + return nil, resources.ActorRef{}, "worker has no actor assignment", nil } + actor, err := s.store.GetActor(ctx, actorRef) if err != nil { if errors.Is(err, store.ErrNotFound) { - return nil, resources.ActorRef{}, deny("assigned actor not found") + return nil, resources.ActorRef{}, "assigned actor not found", nil } slog.ErrorContext(ctx, "ActorIdentity: failed to read actor", slog.Any("actor", actorRef), slog.Any("err", err)) - return nil, resources.ActorRef{}, status.Error(codes.Internal, "failed to look up actor") + return nil, resources.ActorRef{}, "", status.Error(codes.Internal, "failed to look up actor") } - // Deletion is only entered from SUSPENDED or CRASHED, both of which - // have already released the worker, so the assignment check below would - // reject this too. It is kept for better visibility and logging. if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_DELETING { slog.WarnContext(ctx, "ActorIdentity refused: actor is being deleted", slog.Any("actor", actorRef)) - return nil, resources.ActorRef{}, status.Error(codes.FailedPrecondition, "actor is being deleted") + return nil, resources.ActorRef{}, "", status.Error(codes.FailedPrecondition, "actor is being deleted") } - // An actor placed on a worker always carries its placement fields. Missing - // placement is a control-plane bug rather than a client error, so it is not - // folded into deny(). assignment := actor.GetStatus().GetWorkerAssignment() if assignment == nil { slog.ErrorContext(ctx, "ActorIdentity: running actor has no worker assignment", slog.Any("actor", actorRef)) - return nil, resources.ActorRef{}, status.Error(codes.FailedPrecondition, "actor has no worker assigned") + return nil, resources.ActorRef{}, "", status.Error(codes.FailedPrecondition, "actor has no worker assigned") } if worker.GetStatus().GetAssignment().GetActorUid() != actor.GetMetadata().GetUid() { - return nil, resources.ActorRef{}, deny("worker is no longer assigned to this actor incarnation", slog.Any("actor", actorRef)) + return nil, resources.ActorRef{}, "worker is no longer assigned to this actor incarnation", nil } if assignment.GetWorker().GetName() != worker.GetMetadata().GetName() { - return nil, resources.ActorRef{}, deny("actor no longer points to the requesting worker", slog.Any("actor", actorRef)) + return nil, resources.ActorRef{}, "actor no longer points to the requesting worker", nil } - return actor, actorRef, nil + return actor, actorRef, "", nil }