e80875ac297dcec023aa586fb2e6c0502ced0293
[openafs.git] / src / ubik / remote.c
1 /*
2  * Copyright 2000, International Business Machines Corporation and others.
3  * All Rights Reserved.
4  * 
5  * This software has been released under the terms of the IBM Public
6  * License.  For details, see the LICENSE file in the top-level source
7  * directory or online at http://www.openafs.org/dl/license10.html
8  */
9
10 #include <afsconfig.h>
11 #include <afs/param.h>
12
13 RCSID
14     ("$Header$");
15
16 #include <sys/types.h>
17 #ifdef AFS_NT40_ENV
18 #include <winsock2.h>
19 #include <fcntl.h>
20 #else
21 #include <sys/file.h>
22 #include <netinet/in.h>
23 #endif
24 #include <string.h>
25 #include <lock.h>
26 #include <rx/xdr.h>
27 #include <rx/rx.h>
28 #include <errno.h>
29 #include <afs/afsutil.h>
30
31 #define UBIK_INTERNALS
32 #include "ubik.h"
33 #include "ubik_int.h"
34 int (*ubik_CheckRXSecurityProc) ();
35 char *ubik_CheckRXSecurityRock;
36 void printServerInfo();
37
38 /* routines for handling requests remotely-submitted by the sync site.  These are
39     only write transactions (we don't propagate read trans), and there is at most one
40     write transaction extant at any one time.
41 */
42
43 struct ubik_trans *ubik_currentTrans = 0;
44
45
46 ubik_CheckAuth(acall)
47      register struct rx_call *acall;
48 {
49     register afs_int32 code;
50     if (ubik_CheckRXSecurityProc) {
51         code = (*ubik_CheckRXSecurityProc) (ubik_CheckRXSecurityRock, acall);
52         return code;
53     } else
54         return 0;
55 }
56
57 /* the rest of these guys handle remote execution of write
58  * transactions: this is the code executed on the other servers when a
59  * sync site is executing a write transaction.
60  */
61 afs_int32
62 SDISK_Begin(rxcall, atid)
63      register struct rx_call *rxcall;
64      struct ubik_tid *atid;
65 {
66     register afs_int32 code;
67
68     if ((code = ubik_CheckAuth(rxcall))) {
69         return code;
70     }
71     DBHOLD(ubik_dbase);
72     urecovery_CheckTid(atid);
73     if (ubik_currentTrans) {
74         /* If the thread is not waiting for lock - ok to end it */
75 #if !defined(UBIK_PAUSE)
76         if (ubik_currentTrans->locktype != LOCKWAIT) {
77 #endif /* UBIK_PAUSE */
78             udisk_end(ubik_currentTrans);
79 #if !defined(UBIK_PAUSE)
80         }
81 #endif /* UBIK_PAUSE */
82         ubik_currentTrans = (struct ubik_trans *)0;
83     }
84     code = udisk_begin(ubik_dbase, UBIK_WRITETRANS, &ubik_currentTrans);
85     if (!code && ubik_currentTrans) {
86         /* label this trans with the right trans id */
87         ubik_currentTrans->tid.epoch = atid->epoch;
88         ubik_currentTrans->tid.counter = atid->counter;
89     }
90     DBRELE(ubik_dbase);
91     return code;
92 }
93
94
95 afs_int32
96 SDISK_Commit(rxcall, atid)
97      register struct rx_call *rxcall;
98      struct ubik_tid *atid;
99 {
100     register afs_int32 code;
101     register struct ubik_dbase *dbase;
102
103     if ((code = ubik_CheckAuth(rxcall))) {
104         return code;
105     }
106
107     if (!ubik_currentTrans) {
108         return USYNC;
109     }
110     /*
111      * sanity check to make sure only write trans appear here
112      */
113     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
114         return UBADTYPE;
115     }
116
117     dbase = ubik_currentTrans->dbase;
118     DBHOLD(dbase);
119     urecovery_CheckTid(atid);
120     if (!ubik_currentTrans) {
121         DBRELE(dbase);
122         return USYNC;
123     }
124
125     code = udisk_commit(ubik_currentTrans);
126     if (code == 0) {
127         /* sync site should now match */
128         ubik_dbVersion = ubik_dbase->version;
129     }
130     DBRELE(dbase);
131     return code;
132 }
133
134 afs_int32
135 SDISK_ReleaseLocks(rxcall, atid)
136      register struct rx_call *rxcall;
137      struct ubik_tid *atid;
138 {
139     register struct ubik_dbase *dbase;
140     register afs_int32 code;
141
142     if ((code = ubik_CheckAuth(rxcall))) {
143         return code;
144     }
145
146     if (!ubik_currentTrans) {
147         return USYNC;
148     }
149     /* sanity check to make sure only write trans appear here */
150     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
151         return UBADTYPE;
152     }
153
154     dbase = ubik_currentTrans->dbase;
155     DBHOLD(dbase);
156     urecovery_CheckTid(atid);
157     if (!ubik_currentTrans) {
158         DBRELE(dbase);
159         return USYNC;
160     }
161
162     /* If the thread is not waiting for lock - ok to end it */
163 #if !defined(UBIK_PAUSE)
164     if (ubik_currentTrans->locktype != LOCKWAIT) {
165 #endif /* UBIK_PAUSE */
166         udisk_end(ubik_currentTrans);
167 #if !defined(UBIK_PAUSE)
168     }
169 #endif /* UBIK_PAUSE */
170     ubik_currentTrans = (struct ubik_trans *)0;
171     DBRELE(dbase);
172     return 0;
173 }
174
175 afs_int32
176 SDISK_Abort(rxcall, atid)
177      register struct rx_call *rxcall;
178      struct ubik_tid *atid;
179 {
180     register afs_int32 code;
181     register struct ubik_dbase *dbase;
182
183     if ((code = ubik_CheckAuth(rxcall))) {
184         return code;
185     }
186
187     if (!ubik_currentTrans) {
188         return USYNC;
189     }
190     /* sanity check to make sure only write trans appear here  */
191     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
192         return UBADTYPE;
193     }
194
195     dbase = ubik_currentTrans->dbase;
196     DBHOLD(dbase);
197     urecovery_CheckTid(atid);
198     if (!ubik_currentTrans) {
199         DBRELE(dbase);
200         return USYNC;
201     }
202
203     code = udisk_abort(ubik_currentTrans);
204     /* If the thread is not waiting for lock - ok to end it */
205 #if !defined(UBIK_PAUSE)
206     if (ubik_currentTrans->locktype != LOCKWAIT) {
207 #endif /* UBIK_PAUSE */
208         udisk_end(ubik_currentTrans);
209 #if !defined(UBIK_PAUSE)
210     }
211 #endif /* UBIK_PAUSE */
212     ubik_currentTrans = (struct ubik_trans *)0;
213     DBRELE(dbase);
214     return code;
215 }
216
217 afs_int32
218 SDISK_Lock(rxcall, atid, afile, apos, alen, atype)
219      register struct rx_call *rxcall;
220      struct ubik_tid *atid;
221      afs_int32 afile, apos, alen, atype;        /* apos and alen are not used */
222 {
223     register afs_int32 code;
224     register struct ubik_dbase *dbase;
225     struct ubik_trans *ubik_thisTrans;
226
227     if ((code = ubik_CheckAuth(rxcall))) {
228         return code;
229     }
230     if (!ubik_currentTrans) {
231         return USYNC;
232     }
233     /* sanity check to make sure only write trans appear here */
234     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
235         return UBADTYPE;
236     }
237     if (alen != 1) {
238         return UBADLOCK;
239     }
240     dbase = ubik_currentTrans->dbase;
241     DBHOLD(dbase);
242     urecovery_CheckTid(atid);
243     if (!ubik_currentTrans) {
244         DBRELE(dbase);
245         return USYNC;
246     }
247
248     ubik_thisTrans = ubik_currentTrans;
249     code = ulock_getLock(ubik_currentTrans, atype, 1);
250
251     /* While waiting, the transaction may have been ended/
252      * aborted from under us (urecovery_CheckTid). In that
253      * case, end the transaction here.
254      */
255     if (!code && (ubik_currentTrans != ubik_thisTrans)) {
256         udisk_end(ubik_thisTrans);
257         code = USYNC;
258     }
259
260     DBRELE(dbase);
261     return code;
262 }
263
264 /* Write a vector of data */
265 afs_int32
266 SDISK_WriteV(rxcall, atid, io_vector, io_buffer)
267      register struct rx_call *rxcall;
268      struct ubik_tid *atid;
269      iovec_wrt *io_vector;
270      iovec_buf *io_buffer;
271 {
272     afs_int32 code, i, offset;
273     struct ubik_dbase *dbase;
274     struct ubik_iovec *iovec;
275     char *iobuf;
276
277     if ((code = ubik_CheckAuth(rxcall))) {
278         return code;
279     }
280     if (!ubik_currentTrans) {
281         return USYNC;
282     }
283     /* sanity check to make sure only write trans appear here */
284     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
285         return UBADTYPE;
286     }
287
288     dbase = ubik_currentTrans->dbase;
289     DBHOLD(dbase);
290     urecovery_CheckTid(atid);
291     if (!ubik_currentTrans) {
292         DBRELE(dbase);
293         return USYNC;
294     }
295
296     iovec = (struct ubik_iovec *)io_vector->iovec_wrt_val;
297     iobuf = (char *)io_buffer->iovec_buf_val;
298     for (i = 0, offset = 0; i < io_vector->iovec_wrt_len; i++) {
299         /* Sanity check for going off end of buffer */
300         if ((offset + iovec[i].length) > io_buffer->iovec_buf_len) {
301             code = UINTERNAL;
302         } else {
303             code =
304                 udisk_write(ubik_currentTrans, iovec[i].file, &iobuf[offset],
305                             iovec[i].position, iovec[i].length);
306         }
307         if (code)
308             break;
309
310         offset += iovec[i].length;
311     }
312
313     DBRELE(dbase);
314     return code;
315 }
316
317 afs_int32
318 SDISK_Write(rxcall, atid, afile, apos, adata)
319      register struct rx_call *rxcall;
320      struct ubik_tid *atid;
321      afs_int32 afile, apos;
322      register bulkdata *adata;
323 {
324     register afs_int32 code;
325     register struct ubik_dbase *dbase;
326
327     if ((code = ubik_CheckAuth(rxcall))) {
328         return code;
329     }
330     if (!ubik_currentTrans) {
331         return USYNC;
332     }
333     /* sanity check to make sure only write trans appear here */
334     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
335         return UBADTYPE;
336     }
337
338     dbase = ubik_currentTrans->dbase;
339     DBHOLD(dbase);
340     urecovery_CheckTid(atid);
341     if (!ubik_currentTrans) {
342         DBRELE(dbase);
343         return USYNC;
344     }
345     code =
346         udisk_write(ubik_currentTrans, afile, adata->bulkdata_val, apos,
347                     adata->bulkdata_len);
348     DBRELE(dbase);
349     return code;
350 }
351
352 afs_int32
353 SDISK_Truncate(rxcall, atid, afile, alen)
354      register struct rx_call *rxcall;
355      struct ubik_tid *atid;
356      afs_int32 afile;
357      afs_int32 alen;
358 {
359     register afs_int32 code;
360     register struct ubik_dbase *dbase;
361
362     if ((code = ubik_CheckAuth(rxcall))) {
363         return code;
364     }
365     if (!ubik_currentTrans) {
366         return USYNC;
367     }
368     /* sanity check to make sure only write trans appear here */
369     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
370         return UBADTYPE;
371     }
372
373     dbase = ubik_currentTrans->dbase;
374     DBHOLD(dbase);
375     urecovery_CheckTid(atid);
376     if (!ubik_currentTrans) {
377         DBRELE(dbase);
378         return USYNC;
379     }
380     code = udisk_truncate(ubik_currentTrans, afile, alen);
381     DBRELE(dbase);
382     return code;
383 }
384
385 afs_int32
386 SDISK_GetVersion(rxcall, aversion)
387      register struct rx_call *rxcall;
388      register struct ubik_version *aversion;
389 {
390     register afs_int32 code;
391
392     if ((code = ubik_CheckAuth(rxcall))) {
393         return code;
394     }
395
396     /*
397      * If we are the sync site, recovery shouldn't be running on any
398      * other site. We shouldn't be getting this RPC as long as we are
399      * the sync site.  To prevent any unforseen activity, we should
400      * reject this RPC until we have recognized that we are not the
401      * sync site anymore, and/or if we have any pending WRITE
402      * transactions that have to complete. This way we can be assured
403      * that this RPC would not block any pending transactions that
404      * should either fail or pass. If we have recognized the fact that
405      * we are not the sync site any more, all write transactions would
406      * fail with UNOQUORUM anyway.
407      */
408     if (ubeacon_AmSyncSite()) {
409         return UDEADLOCK;
410     }
411
412     DBHOLD(ubik_dbase);
413     code = (*ubik_dbase->getlabel) (ubik_dbase, 0, aversion);
414     DBRELE(ubik_dbase);
415     if (code) {
416         /* tell other side there's no dbase */
417         aversion->epoch = 0;
418         aversion->counter = 0;
419     }
420     return 0;
421 }
422
423 afs_int32
424 SDISK_GetFile(rxcall, file, version)
425      register struct rx_call *rxcall;
426      register afs_int32 file;
427      struct ubik_version *version;
428 {
429     register afs_int32 code;
430     register struct ubik_dbase *dbase;
431     register afs_int32 offset;
432     struct ubik_stat ubikstat;
433     char tbuffer[256];
434     afs_int32 tlen;
435     afs_int32 length;
436
437     if ((code = ubik_CheckAuth(rxcall))) {
438         return code;
439     }
440 /* temporarily disabled because it causes problems for migration tool.  Hey, it's just
441  * a sanity check, anyway. 
442     if (ubeacon_AmSyncSite()) {
443       return UDEADLOCK;
444     }
445 */
446     dbase = ubik_dbase;
447     DBHOLD(dbase);
448     code = (*dbase->stat) (dbase, file, &ubikstat);
449     if (code < 0) {
450         DBRELE(dbase);
451         return code;
452     }
453     length = ubikstat.size;
454     tlen = htonl(length);
455     code = rx_Write(rxcall, (char *)&tlen, sizeof(afs_int32));
456     if (code != sizeof(afs_int32)) {
457         DBRELE(dbase);
458         ubik_dprint("Rx-write length error=%d\n", code);
459         return BULK_ERROR;
460     }
461     offset = 0;
462     while (length > 0) {
463         tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
464         code = (*dbase->read) (dbase, file, tbuffer, offset, tlen);
465         if (code != tlen) {
466             DBRELE(dbase);
467             ubik_dprint("read failed error=%d\n", code);
468             return UIOERROR;
469         }
470         code = rx_Write(rxcall, tbuffer, tlen);
471         if (code != tlen) {
472             DBRELE(dbase);
473             ubik_dprint("Rx-write length error=%d\n", code);
474             return BULK_ERROR;
475         }
476         length -= tlen;
477         offset += tlen;
478     }
479     code = (*dbase->getlabel) (dbase, file, version);   /* return the dbase, too */
480     DBRELE(dbase);
481     return code;
482 }
483
484 afs_int32
485 SDISK_SendFile(rxcall, file, length, avers)
486      register struct rx_call *rxcall;
487      afs_int32 file;
488      afs_int32 length;
489      struct ubik_version *avers;
490 {
491     register afs_int32 code;
492     register struct ubik_dbase *dbase;
493     char tbuffer[1024];
494     afs_int32 offset;
495     struct ubik_version tversion;
496     register int tlen;
497     struct rx_peer *tpeer;
498     struct rx_connection *tconn;
499     afs_uint32 otherHost;
500 #ifndef OLD_URECOVERY
501     char pbuffer[1028];
502     int flen, fd = -1;
503 #endif
504
505     /* send the file back to the requester */
506
507     if ((code = ubik_CheckAuth(rxcall))) {
508         goto failed;
509     }
510
511     /* next, we do a sanity check to see if the guy sending us the database is
512      * the guy we think is the sync site.  It turns out that we might not have
513      * decided yet that someone's the sync site, but they could have enough
514      * votes from others to be sync site anyway, and could send us the database
515      * in advance of getting our votes.  This is fine, what we're really trying
516      * to check is that some authenticated bogon isn't sending a random database
517      * into another configuration.  This could happen on a bad configuration
518      * screwup.  Thus, we only object if we're sure we know who the sync site
519      * is, and it ain't the guy talking to us.
520      */
521     offset = uvote_GetSyncSite();
522     tconn = rx_ConnectionOf(rxcall);
523     tpeer = rx_PeerOf(tconn);
524     otherHost = ubikGetPrimaryInterfaceAddr(rx_HostOf(tpeer));
525     if (offset && offset != otherHost) {
526         /* we *know* this is the wrong guy */
527         code = USYNC;
528         goto failed;
529     }
530
531     dbase = ubik_dbase;
532     DBHOLD(dbase);
533
534     /* abort any active trans that may scribble over the database */
535     urecovery_AbortAll(dbase);
536
537     ubik_print("Ubik: Synchronize database with server %s\n",
538                afs_inet_ntoa(otherHost));
539
540     offset = 0;
541 #ifdef OLD_URECOVERY
542     (*dbase->truncate) (dbase, file, 0);        /* truncate first */
543     tversion.epoch = 0;         /* start off by labelling in-transit db as invalid */
544     tversion.counter = 0;
545     (*dbase->setlabel) (dbase, file, &tversion);        /* setlabel does sync */
546 #else
547     flen = length;
548     afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB0.TMP", ubik_dbase->pathName);
549     fd = open(pbuffer, O_CREAT | O_RDWR | O_TRUNC, 0600);
550     if (fd < 0) {
551         code = errno;
552         goto failed;
553     }
554     code = lseek(fd, HDRSIZE, 0);
555     if (code != HDRSIZE) {
556         close(fd);
557         goto failed;
558     }
559 #endif
560     memcpy(&ubik_dbase->version, &tversion, sizeof(struct ubik_version));
561     while (length > 0) {
562         tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
563         code = rx_Read(rxcall, tbuffer, tlen);
564         if (code != tlen) {
565             DBRELE(dbase);
566             ubik_dprint("Rx-read length error=%d\n", code);
567             code = BULK_ERROR;
568             close(fd);
569             goto failed;
570         }
571 #ifdef OLD_URECOVERY
572         code = (*dbase->write) (dbase, file, tbuffer, offset, tlen);
573 #else
574         code = write(fd, tbuffer, tlen);
575 #endif
576         if (code != tlen) {
577             DBRELE(dbase);
578             ubik_dprint("write failed error=%d\n", code);
579             code = UIOERROR;
580             close(fd);
581             goto failed;
582         }
583         offset += tlen;
584         length -= tlen;
585     }
586 #ifndef OLD_URECOVERY
587     code = close(fd);
588     if (code)
589         goto failed;
590 #endif     
591
592     /* sync data first, then write label and resync (resync done by setlabel call).
593      * This way, good label is only on good database. */
594 #ifdef OLD_URECOVERY
595     (*ubik_dbase->sync) (dbase, file);
596 #else
597     afs_snprintf(tbuffer, sizeof(tbuffer), "%s.DB0", ubik_dbase->pathName);
598 #ifdef AFS_NT40_ENV
599     afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB0.OLD", ubik_dbase->pathName);
600     code = unlink(pbuffer);
601     if (!code)
602         code = rename(tbuffer, pbuffer);
603     afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB0.TMP", ubik_dbase->pathName);
604 #endif
605     if (!code) 
606         code = rename(pbuffer, tbuffer);
607     if (!code)
608 #endif
609     code = (*ubik_dbase->setlabel) (dbase, file, avers);
610 #ifndef OLD_URECOVERY
611 #ifdef AFS_NT40_ENV
612     afs_snprintf(pbuffer, sizeof(pbuffer), "%s.DB0.OLD", ubik_dbase->pathName);
613     unlink(pbuffer);
614 #endif
615 #endif
616     memcpy(&ubik_dbase->version, avers, sizeof(struct ubik_version));
617     udisk_Invalidate(dbase, file);      /* new dbase, flush disk buffers */
618     LWP_NoYieldSignal(&dbase->version);
619     DBRELE(dbase);
620   failed:
621     if (code) {
622 #ifndef OLD_URECOVERY
623         unlink(pbuffer);
624 #endif
625         ubik_print
626             ("Ubik: Synchronize database with server %s failed (error = %d)\n",
627              afs_inet_ntoa(otherHost), code);
628     } else {
629         ubik_print("Ubik: Synchronize database completed\n");
630     }
631     return code;
632 }
633
634
635 afs_int32
636 SDISK_Probe(rxcall)
637      register struct rx_call *rxcall;
638 {
639     return 0;
640 }
641
642 /*
643 * Update remote machines addresses in my server list
644 * Send back my addresses to caller of this RPC
645 * Returns zero on success, else 1.
646 */
647 afs_int32
648 SDISK_UpdateInterfaceAddr(rxcall, inAddr, outAddr)
649      register struct rx_call *rxcall;
650      UbikInterfaceAddr *inAddr, *outAddr;
651 {
652     struct ubik_server *ts, *tmp;
653     afs_uint32 remoteAddr;      /* in net byte order */
654     int i, j, found = 0, probableMatch = 0;
655
656     /* copy the output parameters */
657     for (i = 0; i < UBIK_MAX_INTERFACE_ADDR; i++)
658         outAddr->hostAddr[i] = ntohl(ubik_host[i]);
659
660     remoteAddr = htonl(inAddr->hostAddr[0]);
661     for (ts = ubik_servers; ts; ts = ts->next)
662         if (ts->addr[0] == remoteAddr) {        /* both in net byte order */
663             probableMatch = 1;
664             break;
665         }
666
667     if (probableMatch) {
668         /* verify that all addresses in the incoming RPC are
669          ** not part of other server entries in my CellServDB
670          */
671         for (i = 0; !found && (i < UBIK_MAX_INTERFACE_ADDR)
672              && inAddr->hostAddr[i]; i++) {
673             remoteAddr = htonl(inAddr->hostAddr[i]);
674             for (tmp = ubik_servers; (!found && tmp); tmp = tmp->next) {
675                 if (ts == tmp)  /* this is my server */
676                     continue;
677                 for (j = 0; (j < UBIK_MAX_INTERFACE_ADDR) && tmp->addr[j];
678                      j++)
679                     if (remoteAddr == tmp->addr[j]) {
680                         found = 1;
681                         break;
682                     }
683             }
684         }
685     }
686
687     /* if (probableMatch) */
688     /* inconsistent addresses in CellServDB */
689     if (!probableMatch || found) {
690         ubik_print("Inconsistent Cell Info from server: ");
691         for (i = 0; i < UBIK_MAX_INTERFACE_ADDR && inAddr->hostAddr[i]; i++)
692             ubik_print("%s ", afs_inet_ntoa(htonl(inAddr->hostAddr[i])));
693         ubik_print("\n");
694         fflush(stdout);
695         fflush(stderr);
696         printServerInfo();
697         return UBADHOST;
698     }
699
700     /* update our data structures */
701     for (i = 1; i < UBIK_MAX_INTERFACE_ADDR; i++)
702         ts->addr[i] = htonl(inAddr->hostAddr[i]);
703
704     ubik_print("ubik: A Remote Server has addresses: ");
705     for (i = 0; i < UBIK_MAX_INTERFACE_ADDR && ts->addr[i]; i++)
706         ubik_print("%s ", afs_inet_ntoa(ts->addr[i]));
707     ubik_print("\n");
708
709     return 0;
710 }
711
712 void
713 printServerInfo()
714 {
715     struct ubik_server *ts;
716     int i, j = 1;
717
718     ubik_print("Local CellServDB:");
719     for (ts = ubik_servers; ts; ts = ts->next, j++) {
720         ubik_print("Server %d: ", j);
721         for (i = 0; (i < UBIK_MAX_INTERFACE_ADDR) && ts->addr[i]; i++)
722             ubik_print("%s ", afs_inet_ntoa(ts->addr[i]));
723     }
724     ubik_print("\n");
725 }
726
727 afs_int32
728 SDISK_SetVersion(rxcall, atid, oldversionp, newversionp)
729      struct rx_call *rxcall;
730      struct ubik_tid *atid;
731      struct ubik_version *oldversionp;
732      struct ubik_version *newversionp;
733 {
734     afs_int32 code = 0;
735     struct ubik_dbase *dbase;
736
737     if ((code = ubik_CheckAuth(rxcall))) {
738         return (code);
739     }
740
741     if (!ubik_currentTrans) {
742         return USYNC;
743     }
744     /* sanity check to make sure only write trans appear here */
745     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
746         return UBADTYPE;
747     }
748
749     /* Should not get this for the sync site */
750     if (ubeacon_AmSyncSite()) {
751         return UDEADLOCK;
752     }
753
754     dbase = ubik_currentTrans->dbase;
755     DBHOLD(dbase);
756     urecovery_CheckTid(atid);
757     if (!ubik_currentTrans) {
758         DBRELE(dbase);
759         return USYNC;
760     }
761
762     /* Set the label if its version matches the sync-site's */
763     if ((oldversionp->epoch == ubik_dbVersion.epoch)
764         && (oldversionp->counter == ubik_dbVersion.counter)) {
765         code = (*dbase->setlabel) (ubik_dbase, 0, newversionp);
766         if (!code) {
767             ubik_dbase->version = *newversionp;
768             ubik_dbVersion = *newversionp;
769         }
770     } else {
771         code = USYNC;
772     }
773
774     DBRELE(dbase);
775     return code;
776 }