source4/wrepl_server/wrepl_out_helpers.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    WINS Replication server
   5
   6    Copyright (C) Stefan Metzmacher      2005
   7
   8    This program is free software; you can redistribute it and/or modify
   9    it under the terms of the GNU General Public License as published by
  10    the Free Software Foundation; either version 2 of the License, or
  11    (at your option) any later version.
  12
  13    This program is distributed in the hope that it will be useful,
  14    but WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16    GNU General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21 */
  22
  23 #include "includes.h"
  24 #include "lib/events/events.h"
  25 #include "lib/socket/socket.h"
  26 #include "smbd/service_task.h"
  27 #include "smbd/service_stream.h"
  28 #include "librpc/gen_ndr/winsrepl.h"
  29 #include "wrepl_server/wrepl_server.h"
  30 #include "nbt_server/wins/winsdb.h"
  31 #include "libcli/composite/composite.h"
  32 #include "libcli/wrepl/winsrepl.h"
  33
  34 enum wreplsrv_out_connect_stage {
  35         WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
  36         WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
  37         WREPLSRV_OUT_CONNECT_STAGE_DONE
  38 };
  39
  40 struct wreplsrv_out_connect_state {
  41         enum wreplsrv_out_connect_stage stage;
  42         struct composite_context *c;
  43         struct wrepl_request *req;
  44         struct composite_context *c_req;
  45         struct wrepl_associate assoc_io;
  46         enum winsrepl_partner_type type;
  47         struct wreplsrv_out_connection *wreplconn;
  48 };
  49
  50 static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
  51 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
  52
  53 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
  54 {
  55         NTSTATUS status;
  56
  57         status = wrepl_connect_recv(state->c_req);
  58         NT_STATUS_NOT_OK_RETURN(status);
  59
  60         state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
  61         NT_STATUS_HAVE_NO_MEMORY(state->req);
  62
  63         state->req->async.fn            = wreplsrv_out_connect_handler_req;
  64         state->req->async.private       = state;
  65
  66         state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
  67
  68         return NT_STATUS_OK;
  69 }
  70
  71 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
  72 {
  73         NTSTATUS status;
  74
  75         status = wrepl_associate_recv(state->req, &state->assoc_io);
  76         NT_STATUS_NOT_OK_RETURN(status);
  77
  78         state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
  79
  80         if (state->type == WINSREPL_PARTNER_PUSH) {
  81                 state->wreplconn->partner->push.wreplconn = state->wreplconn;
  82                 talloc_steal(state->wreplconn->partner, state->wreplconn);
  83         } else if (state->type == WINSREPL_PARTNER_PULL) {
  84                 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
  85                 talloc_steal(state->wreplconn->partner, state->wreplconn);
  86         }
  87
  88         state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
  89
  90         return NT_STATUS_OK;
  91 }
  92
  93 static void wreplsrv_out_connect_handler(struct wreplsrv_out_connect_state *state)
  94 {
  95         struct composite_context *c = state->c;
  96
  97         switch (state->stage) {
  98         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
  99                 c->status = wreplsrv_out_connect_wait_socket(state);
 100                 break;
 101         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
 102                 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
 103                 c->state  = COMPOSITE_STATE_DONE;
 104                 break;
 105         case WREPLSRV_OUT_CONNECT_STAGE_DONE:
 106                 c->status = NT_STATUS_INTERNAL_ERROR;
 107         }
 108
 109         if (!NT_STATUS_IS_OK(c->status)) {
 110                 c->state = COMPOSITE_STATE_ERROR;
 111         }
 112
 113         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 114                 c->async.fn(c);
 115         }
 116 }
 117
 118 static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
 119 {
 120         struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
 121                                                    struct wreplsrv_out_connect_state);
 122         wreplsrv_out_connect_handler(state);
 123         return;
 124 }
 125
 126 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
 127 {
 128         struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
 129                                                    struct wreplsrv_out_connect_state);
 130         wreplsrv_out_connect_handler(state);
 131         return;
 132 }
 133
 134 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
 135                                                            enum winsrepl_partner_type type,
 136                                                            struct wreplsrv_out_connection *wreplconn)
 137 {
 138         struct composite_context *c = NULL;
 139         struct wreplsrv_service *service = partner->service;
 140         struct wreplsrv_out_connect_state *state = NULL;
 141         struct wreplsrv_out_connection **wreplconnp = &wreplconn;
 142         BOOL cached_connection = False;
 143
 144         c = talloc_zero(partner, struct composite_context);
 145         if (!c) goto failed;
 146
 147         state = talloc_zero(c, struct wreplsrv_out_connect_state);
 148         if (!state) goto failed;
 149         state->c        = c;
 150         state->type     = type;
 151
 152         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 153         c->event_ctx    = service->task->event_ctx;
 154         c->private_data = state;
 155
 156         if (type == WINSREPL_PARTNER_PUSH) {
 157                 cached_connection       = True;
 158                 wreplconn               = partner->push.wreplconn;
 159                 wreplconnp              = &partner->push.wreplconn;
 160         } else if (type == WINSREPL_PARTNER_PULL) {
 161                 cached_connection       = True;
 162                 wreplconn               = partner->pull.wreplconn;
 163                 wreplconnp              = &partner->pull.wreplconn;
 164         }
 165
 166         /* we have a connection already, so use it */
 167         if (wreplconn) {
 168                 if (!wreplconn->sock->dead) {
 169                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
 170                         state->wreplconn= wreplconn;
 171                         composite_done(c);
 172                         return c;
 173                 } else if (!cached_connection) {
 174                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
 175                         state->wreplconn= NULL;
 176                         composite_done(c);
 177                         return c;
 178                 } else {
 179                         talloc_free(wreplconn);
 180                         *wreplconnp = NULL;
 181                 }
 182         }
 183
 184         wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
 185         if (!wreplconn) goto failed;
 186
 187         wreplconn->service      = service;
 188         wreplconn->partner      = partner;
 189         wreplconn->sock         = wrepl_socket_init(wreplconn, service->task->event_ctx);
 190         if (!wreplconn->sock) goto failed;
 191
 192         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
 193         state->wreplconn= wreplconn;
 194         state->c_req    = wrepl_connect_send(wreplconn->sock,
 195                                              partner->our_address,
 196                                              partner->address);
 197         if (!state->c_req) goto failed;
 198
 199         state->c_req->async.fn                  = wreplsrv_out_connect_handler_creq;
 200         state->c_req->async.private_data        = state;
 201
 202         return c;
 203 failed:
 204         talloc_free(c);
 205         return NULL;
 206 }
 207
 208 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 209                                           struct wreplsrv_out_connection **wreplconn)
 210 {
 211         NTSTATUS status;
 212
 213         status = composite_wait(c);
 214
 215         if (NT_STATUS_IS_OK(status)) {
 216                 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
 217                                                            struct wreplsrv_out_connect_state);
 218                 if (state->wreplconn) {
 219                         *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
 220                         if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
 221                 } else {
 222                         status = NT_STATUS_INVALID_CONNECTION;
 223                 }
 224         }
 225
 226         talloc_free(c);
 227         return status;
 228
 229 }
 230
 231 struct wreplsrv_pull_table_io {
 232         struct {
 233                 struct wreplsrv_partner *partner;
 234                 uint32_t num_owners;
 235                 struct wrepl_wins_owner *owners;
 236         } in;
 237         struct {
 238                 uint32_t num_owners;
 239                 struct wrepl_wins_owner *owners;
 240         } out;
 241 };
 242
 243 enum wreplsrv_pull_table_stage {
 244         WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
 245         WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
 246         WREPLSRV_PULL_TABLE_STAGE_DONE
 247 };
 248
 249 struct wreplsrv_pull_table_state {
 250         enum wreplsrv_pull_table_stage stage;
 251         struct composite_context *c;
 252         struct wrepl_request *req;
 253         struct wrepl_pull_table table_io;
 254         struct wreplsrv_pull_table_io *io;
 255         struct composite_context *creq;
 256         struct wreplsrv_out_connection *wreplconn;
 257 };
 258
 259 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
 260
 261 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
 262 {
 263         NTSTATUS status;
 264
 265         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 266         NT_STATUS_NOT_OK_RETURN(status);
 267
 268         state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
 269         state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
 270         NT_STATUS_HAVE_NO_MEMORY(state->req);
 271
 272         state->req->async.fn            = wreplsrv_pull_table_handler_req;
 273         state->req->async.private       = state;
 274
 275         state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
 276
 277         return NT_STATUS_OK;
 278 }
 279
 280 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
 281 {
 282         NTSTATUS status;
 283
 284         status = wrepl_pull_table_recv(state->req, state, &state->table_io);
 285         NT_STATUS_NOT_OK_RETURN(status);
 286
 287         state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
 288
 289         return NT_STATUS_OK;
 290 }
 291
 292 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
 293 {
 294         struct composite_context *c = state->c;
 295
 296         switch (state->stage) {
 297         case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
 298                 c->status = wreplsrv_pull_table_wait_connection(state);
 299                 break;
 300         case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
 301                 c->status = wreplsrv_pull_table_wait_table_reply(state);
 302                 c->state  = COMPOSITE_STATE_DONE;
 303                 break;
 304         case WREPLSRV_PULL_TABLE_STAGE_DONE:
 305                 c->status = NT_STATUS_INTERNAL_ERROR;
 306         }
 307
 308         if (!NT_STATUS_IS_OK(c->status)) {
 309                 c->state = COMPOSITE_STATE_ERROR;
 310         }
 311
 312         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 313                 c->async.fn(c);
 314         }
 315 }
 316
 317 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
 318 {
 319         struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
 320                                                   struct wreplsrv_pull_table_state);
 321         wreplsrv_pull_table_handler(state);
 322         return;
 323 }
 324
 325 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
 326 {
 327         struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
 328                                                   struct wreplsrv_pull_table_state);
 329         wreplsrv_pull_table_handler(state);
 330         return;
 331 }
 332
 333 static struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
 334 {
 335         struct composite_context *c = NULL;
 336         struct wreplsrv_service *service = io->in.partner->service;
 337         struct wreplsrv_pull_table_state *state = NULL;
 338
 339         c = talloc_zero(mem_ctx, struct composite_context);
 340         if (!c) goto failed;
 341
 342         state = talloc_zero(c, struct wreplsrv_pull_table_state);
 343         if (!state) goto failed;
 344         state->c        = c;
 345         state->io       = io;
 346
 347         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 348         c->event_ctx    = service->task->event_ctx;
 349         c->private_data = state;
 350
 351         if (io->in.num_owners) {
 352                 state->table_io.out.num_partners        = io->in.num_owners;
 353                 state->table_io.out.partners            = io->in.owners;
 354                 state->stage                            = WREPLSRV_PULL_TABLE_STAGE_DONE;
 355                 composite_done(c);
 356                 return c;
 357         }
 358
 359         state->stage    = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
 360         state->creq     = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
 361         if (!state->creq) goto failed;
 362
 363         state->creq->async.fn           = wreplsrv_pull_table_handler_creq;
 364         state->creq->async.private_data = state;
 365
 366         return c;
 367 failed:
 368         talloc_free(c);
 369         return NULL;
 370 }
 371
 372 static NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 373                                          struct wreplsrv_pull_table_io *io)
 374 {
 375         NTSTATUS status;
 376
 377         status = composite_wait(c);
 378
 379         if (NT_STATUS_IS_OK(status)) {
 380                 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
 381                                                           struct wreplsrv_pull_table_state);
 382                 io->out.num_owners      = state->table_io.out.num_partners;
 383                 io->out.owners          = state->table_io.out.partners;
 384                 talloc_reference(mem_ctx, state->table_io.out.partners);
 385         }
 386
 387         talloc_free(c);
 388         return status;
 389 }
 390
 391 enum wreplsrv_pull_names_stage {
 392         WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION,
 393         WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY,
 394         WREPLSRV_PULL_NAMES_STAGE_DONE
 395 };
 396
 397 struct wreplsrv_pull_names_state {
 398         enum wreplsrv_pull_names_stage stage;
 399         struct composite_context *c;
 400         struct wrepl_request *req;
 401         struct wrepl_pull_names pull_io;
 402         struct wreplsrv_pull_names_io *io;
 403         struct composite_context *creq;
 404         struct wreplsrv_out_connection *wreplconn;
 405 };
 406
 407 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
 408
 409 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
 410 {
 411         NTSTATUS status;
 412
 413         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 414         NT_STATUS_NOT_OK_RETURN(status);
 415
 416         state->pull_io.in.assoc_ctx     = state->wreplconn->assoc_ctx.peer_ctx;
 417         state->pull_io.in.partner       = state->io->in.owner;
 418         state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
 419         NT_STATUS_HAVE_NO_MEMORY(state->req);
 420
 421         state->req->async.fn            = wreplsrv_pull_names_handler_req;
 422         state->req->async.private       = state;
 423
 424         state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
 425
 426         return NT_STATUS_OK;
 427 }
 428
 429 static NTSTATUS wreplsrv_pull_names_wait_send_reply(struct wreplsrv_pull_names_state *state)
 430 {
 431         NTSTATUS status;
 432
 433         status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
 434         NT_STATUS_NOT_OK_RETURN(status);
 435
 436         state->stage = WREPLSRV_PULL_NAMES_STAGE_DONE;
 437
 438         return NT_STATUS_OK;
 439 }
 440
 441 static void wreplsrv_pull_names_handler(struct wreplsrv_pull_names_state *state)
 442 {
 443         struct composite_context *c = state->c;
 444
 445         switch (state->stage) {
 446         case WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION:
 447                 c->status = wreplsrv_pull_names_wait_connection(state);
 448                 break;
 449         case WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY:
 450                 c->status = wreplsrv_pull_names_wait_send_reply(state);
 451                 c->state  = COMPOSITE_STATE_DONE;
 452                 break;
 453         case WREPLSRV_PULL_NAMES_STAGE_DONE:
 454                 c->status = NT_STATUS_INTERNAL_ERROR;
 455         }
 456
 457         if (!NT_STATUS_IS_OK(c->status)) {
 458                 c->state = COMPOSITE_STATE_ERROR;
 459         }
 460
 461         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 462                 c->async.fn(c);
 463         }
 464 }
 465
 466 static void wreplsrv_pull_names_handler_creq(struct composite_context *creq)
 467 {
 468         struct wreplsrv_pull_names_state *state = talloc_get_type(creq->async.private_data,
 469                                                   struct wreplsrv_pull_names_state);
 470         wreplsrv_pull_names_handler(state);
 471         return;
 472 }
 473
 474 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
 475 {
 476         struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private,
 477                                                   struct wreplsrv_pull_names_state);
 478         wreplsrv_pull_names_handler(state);
 479         return;
 480 }
 481
 482 struct composite_context *wreplsrv_pull_names_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_names_io *io)
 483 {
 484         struct composite_context *c = NULL;
 485         struct wreplsrv_service *service = io->in.partner->service;
 486         struct wreplsrv_pull_names_state *state = NULL;
 487         enum winsrepl_partner_type partner_type = WINSREPL_PARTNER_PULL;
 488
 489         if (io->in.wreplconn) partner_type = WINSREPL_PARTNER_NONE;
 490
 491         c = talloc_zero(mem_ctx, struct composite_context);
 492         if (!c) goto failed;
 493
 494         state = talloc_zero(c, struct wreplsrv_pull_names_state);
 495         if (!state) goto failed;
 496         state->c        = c;
 497         state->io       = io;
 498
 499         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 500         c->event_ctx    = service->task->event_ctx;
 501         c->private_data = state;
 502
 503         state->stage    = WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION;
 504         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, io->in.wreplconn);
 505         if (!state->creq) goto failed;
 506
 507         state->creq->async.fn           = wreplsrv_pull_names_handler_creq;
 508         state->creq->async.private_data = state;
 509
 510         return c;
 511 failed:
 512         talloc_free(c);
 513         return NULL;
 514 }
 515
 516 NTSTATUS wreplsrv_pull_names_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 517                                   struct wreplsrv_pull_names_io *io)
 518 {
 519         NTSTATUS status;
 520
 521         status = composite_wait(c);
 522
 523         if (NT_STATUS_IS_OK(status)) {
 524                 struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
 525                                                           struct wreplsrv_pull_names_state);
 526                 io->out.num_names       = state->pull_io.out.num_names;
 527                 io->out.names           = state->pull_io.out.names;
 528                 talloc_reference(mem_ctx, state->pull_io.out.names);
 529         }
 530
 531         talloc_free(c);
 532         return status;
 533
 534 }
 535
 536 enum wreplsrv_pull_cycle_stage {
 537         WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY,
 538         WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES,
 539         WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC,
 540         WREPLSRV_PULL_CYCLE_STAGE_DONE
 541 };
 542
 543 struct wreplsrv_pull_cycle_state {
 544         enum wreplsrv_pull_cycle_stage stage;
 545         struct composite_context *c;
 546         struct wreplsrv_pull_cycle_io *io;
 547         struct wreplsrv_pull_table_io table_io;
 548         uint32_t current;
 549         struct wreplsrv_pull_names_io names_io;
 550         struct composite_context *creq;
 551         struct wrepl_associate_stop assoc_stop_io;
 552         struct wrepl_request *req;
 553 };
 554
 555 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
 556 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
 557
 558 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
 559 {
 560         struct wreplsrv_owner *current_owner;
 561         struct wreplsrv_owner *local_owner;
 562         uint32_t i;
 563         uint64_t old_max_version = 0;
 564         BOOL do_pull = False;
 565
 566         for (i=state->current; i < state->table_io.out.num_owners; i++) {
 567                 current_owner = wreplsrv_find_owner(state->io->in.partner->service,
 568                                                     state->io->in.partner->pull.table,
 569                                                     state->table_io.out.owners[i].address);
 570
 571                 local_owner = wreplsrv_find_owner(state->io->in.partner->service,
 572                                                   state->io->in.partner->service->table,
 573                                                   state->table_io.out.owners[i].address);
 574                 /*
 575                  * this means we are ourself the current owner,
 576                  * and we don't want replicate ourself
 577                  */
 578                 if (!current_owner) continue;
 579
 580                 /*
 581                  * this means we don't have any records of this owner
 582                  * so fetch them
 583                  */
 584                 if (!local_owner) {
 585                         do_pull         = True;
 586
 587                         break;
 588                 }
 589
 590                 /*
 591                  * this means the remote partner has some new records of this owner
 592                  * fetch them
 593                  */
 594                 if (current_owner->owner.max_version > local_owner->owner.max_version) {
 595                         do_pull         = True;
 596                         old_max_version = local_owner->owner.max_version;
 597                         break;
 598                 }
 599         }
 600         state->current = i;
 601
 602         if (do_pull) {
 603                 state->names_io.in.partner              = state->io->in.partner;
 604                 state->names_io.in.wreplconn            = state->io->in.wreplconn;
 605                 state->names_io.in.owner                = current_owner->owner;
 606                 state->names_io.in.owner.min_version    = old_max_version + 1;
 607                 state->creq = wreplsrv_pull_names_send(state, &state->names_io);
 608                 NT_STATUS_HAVE_NO_MEMORY(state->creq);
 609
 610                 state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
 611                 state->creq->async.private_data = state;
 612
 613                 return STATUS_MORE_ENTRIES;
 614         }
 615
 616         return NT_STATUS_OK;
 617 }
 618
 619 static NTSTATUS wreplsrv_pull_cycle_next_owner_wrapper(struct wreplsrv_pull_cycle_state *state)
 620 {
 621         NTSTATUS status;
 622
 623         status = wreplsrv_pull_cycle_next_owner_do_work(state);
 624         if (NT_STATUS_IS_OK(status)) {
 625                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
 626         } else if (NT_STATUS_EQUAL(STATUS_MORE_ENTRIES, status)) {
 627                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES;
 628                 status = NT_STATUS_OK;
 629         }
 630
 631         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
 632                 state->assoc_stop_io.in.assoc_ctx       = state->io->in.wreplconn->assoc_ctx.peer_ctx;
 633                 state->assoc_stop_io.in.reason          = 0;
 634                 state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
 635                 NT_STATUS_HAVE_NO_MEMORY(state->req);
 636
 637                 state->req->async.fn            = wreplsrv_pull_cycle_handler_req;
 638                 state->req->async.private       = state;
 639
 640                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
 641         }
 642
 643         return status;
 644 }
 645
 646 static NTSTATUS wreplsrv_pull_cycle_wait_table_reply(struct wreplsrv_pull_cycle_state *state)
 647 {
 648         NTSTATUS status;
 649         uint32_t i;
 650
 651         status = wreplsrv_pull_table_recv(state->creq, state, &state->table_io);
 652         NT_STATUS_NOT_OK_RETURN(status);
 653
 654         /* update partner table */
 655         for (i=0; i < state->table_io.out.num_owners; i++) {
 656                 status = wreplsrv_add_table(state->io->in.partner->service,
 657                                             state->io->in.partner,
 658                                             &state->io->in.partner->pull.table,
 659                                             state->table_io.out.owners[i].address,
 660                                             state->table_io.out.owners[i].max_version);
 661                 NT_STATUS_NOT_OK_RETURN(status);
 662         }
 663
 664         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
 665         NT_STATUS_NOT_OK_RETURN(status);
 666
 667         return status;
 668 }
 669
 670 static NTSTATUS wreplsrv_pull_cycle_apply_records(struct wreplsrv_pull_cycle_state *state)
 671 {
 672         NTSTATUS status;
 673
 674         status = wreplsrv_apply_records(state->io->in.partner, &state->names_io);
 675         NT_STATUS_NOT_OK_RETURN(status);
 676
 677         talloc_free(state->names_io.out.names);
 678         ZERO_STRUCT(state->names_io);
 679
 680         return NT_STATUS_OK;
 681 }
 682
 683 static NTSTATUS wreplsrv_pull_cycle_wait_send_replies(struct wreplsrv_pull_cycle_state *state)
 684 {
 685         NTSTATUS status;
 686
 687         status = wreplsrv_pull_names_recv(state->creq, state, &state->names_io);
 688         NT_STATUS_NOT_OK_RETURN(status);
 689
 690         /*
 691          * TODO: this should maybe an async call,
 692          *       because we may need some network access
 693          *       for conflict resolving
 694          */
 695         status = wreplsrv_pull_cycle_apply_records(state);
 696         NT_STATUS_NOT_OK_RETURN(status);
 697
 698         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
 699         NT_STATUS_NOT_OK_RETURN(status);
 700
 701         return status;
 702 }
 703
 704 static NTSTATUS wreplsrv_pull_cycle_wait_stop_assoc(struct wreplsrv_pull_cycle_state *state)
 705 {
 706         NTSTATUS status;
 707
 708         status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
 709         NT_STATUS_NOT_OK_RETURN(status);
 710
 711         state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
 712
 713         return status;
 714 }
 715
 716 static void wreplsrv_pull_cycle_handler(struct wreplsrv_pull_cycle_state *state)
 717 {
 718         struct composite_context *c = state->c;
 719
 720         switch (state->stage) {
 721         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY:
 722                 c->status = wreplsrv_pull_cycle_wait_table_reply(state);
 723                 break;
 724         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES:
 725                 c->status = wreplsrv_pull_cycle_wait_send_replies(state);
 726                 break;
 727         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC:
 728                 c->status = wreplsrv_pull_cycle_wait_stop_assoc(state);
 729                 break;
 730         case WREPLSRV_PULL_CYCLE_STAGE_DONE:
 731                 c->status = NT_STATUS_INTERNAL_ERROR;
 732         }
 733
 734         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE) {
 735                 c->state  = COMPOSITE_STATE_DONE;
 736         }
 737
 738         if (!NT_STATUS_IS_OK(c->status)) {
 739                 c->state = COMPOSITE_STATE_ERROR;
 740         }
 741
 742         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 743                 c->async.fn(c);
 744         }
 745 }
 746
 747 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq)
 748 {
 749         struct wreplsrv_pull_cycle_state *state = talloc_get_type(creq->async.private_data,
 750                                                   struct wreplsrv_pull_cycle_state);
 751         wreplsrv_pull_cycle_handler(state);
 752         return;
 753 }
 754
 755 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
 756 {
 757         struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private,
 758                                                   struct wreplsrv_pull_cycle_state);
 759         wreplsrv_pull_cycle_handler(state);
 760         return;
 761 }
 762
 763 struct composite_context *wreplsrv_pull_cycle_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_cycle_io *io)
 764 {
 765         struct composite_context *c = NULL;
 766         struct wreplsrv_service *service = io->in.partner->service;
 767         struct wreplsrv_pull_cycle_state *state = NULL;
 768
 769         c = talloc_zero(mem_ctx, struct composite_context);
 770         if (!c) goto failed;
 771
 772         state = talloc_zero(c, struct wreplsrv_pull_cycle_state);
 773         if (!state) goto failed;
 774         state->c        = c;
 775         state->io       = io;
 776
 777         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 778         c->event_ctx    = service->task->event_ctx;
 779         c->private_data = state;
 780
 781         state->stage    = WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY;
 782         state->table_io.in.partner      = io->in.partner;
 783         state->table_io.in.num_owners   = io->in.num_owners;
 784         state->table_io.in.owners       = io->in.owners;
 785         state->creq = wreplsrv_pull_table_send(state, &state->table_io);
 786         if (!state->creq) goto failed;
 787
 788         state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
 789         state->creq->async.private_data = state;
 790
 791         return c;
 792 failed:
 793         talloc_free(c);
 794         return NULL;
 795 }
 796
 797 NTSTATUS wreplsrv_pull_cycle_recv(struct composite_context *c)
 798 {
 799         NTSTATUS status;
 800
 801         status = composite_wait(c);
 802
 803         talloc_free(c);
 804         return status;
 805 }
 806
 807 enum wreplsrv_push_notify_stage {
 808         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
 809         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM,
 810         WREPLSRV_PUSH_NOTIFY_STAGE_DONE
 811 };
 812
 813 struct wreplsrv_push_notify_state {
 814         enum wreplsrv_push_notify_stage stage;
 815         struct composite_context *c;
 816         struct wreplsrv_push_notify_io *io;
 817         enum wrepl_replication_cmd command;
 818         BOOL full_table;
 819         struct wrepl_send_ctrl ctrl;
 820         struct wrepl_request *req;
 821         struct wrepl_packet req_packet;
 822         struct wrepl_packet *rep_packet;
 823         struct composite_context *creq;
 824         struct wreplsrv_out_connection *wreplconn;
 825 };
 826
 827 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
 828 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
 829
 830 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
 831 {
 832         struct wreplsrv_service *service = state->io->in.partner->service;
 833         struct wrepl_packet *req = &state->req_packet;
 834         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
 835         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
 836         struct wreplsrv_in_connection *wrepl_in;
 837         NTSTATUS status;
 838         struct socket_context *sock;
 839         struct packet_context *packet;
 840         uint16_t fde_flags;
 841
 842         /* prepare the outgoing request */
 843         req->opcode     = WREPL_OPCODE_BITS;
 844         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
 845         req->mess_type  = WREPL_REPLICATION;
 846
 847         repl_out->command = state->command;
 848
 849         status = wreplsrv_fill_wrepl_table(service, state, table_out,
 850                                            service->wins_db->local_owner, state->full_table);
 851         NT_STATUS_NOT_OK_RETURN(status);
 852
 853         /* queue the request */
 854         state->req = wrepl_request_send(state->wreplconn->sock, req, NULL);
 855         NT_STATUS_HAVE_NO_MEMORY(state->req);
 856
 857         /*
 858          * now we need to convert the wrepl_socket (client connection)
 859          * into a wreplsrv_in_connection (server connection), because
 860          * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
 861          * message is received by the peer.
 862          */
 863
 864         /* steal the socket_context */
 865         sock = state->wreplconn->sock->sock;
 866         state->wreplconn->sock->sock = NULL;
 867         talloc_steal(state, sock);
 868
 869         /*
 870          * steal the packet_context
 871          * note the request DATA_BLOB we just send on the
 872          * wrepl_socket (client connection) is still unter the
 873          * packet context and will be send to the wire
 874          */
 875         packet = state->wreplconn->sock->packet;
 876         state->wreplconn->sock->packet = NULL;
 877         talloc_steal(state, packet);
 878
 879         /*
 880          * get the fde_flags of the old fde event,
 881          * so that we can later set the same flags to the new one
 882          */
 883         fde_flags = event_get_fd_flags(state->wreplconn->sock->event.fde);
 884
 885         /*
 886          * free the wrepl_socket (client connection)
 887          */
 888         talloc_free(state->wreplconn->sock);
 889         state->wreplconn->sock = NULL;
 890
 891         /*
 892          * now create a wreplsrv_in_connection,
 893          * on which we act as server
 894          *
 895          * NOTE: sock and packet will be stolen by
 896          *       wreplsrv_in_connection_merge()
 897          */
 898         status = wreplsrv_in_connection_merge(state->io->in.partner,
 899                                               sock, packet, &wrepl_in);
 900         NT_STATUS_NOT_OK_RETURN(status);
 901
 902         event_set_fd_flags(wrepl_in->conn->event.fde, fde_flags);
 903
 904         wrepl_in->assoc_ctx.peer_ctx    = state->wreplconn->assoc_ctx.peer_ctx;
 905         wrepl_in->assoc_ctx.our_ctx     = 0;
 906
 907         /* now we can free the wreplsrv_out_connection */
 908         talloc_free(state->wreplconn);
 909         state->wreplconn = NULL;
 910
 911         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
 912
 913         return NT_STATUS_OK;
 914 }
 915
 916 static NTSTATUS wreplsrv_push_notify_inform(struct wreplsrv_push_notify_state *state)
 917 {
 918         struct wreplsrv_service *service = state->io->in.partner->service;
 919         struct wrepl_packet *req = &state->req_packet;
 920         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
 921         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
 922         NTSTATUS status;
 923
 924         req->opcode     = WREPL_OPCODE_BITS;
 925         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
 926         req->mess_type  = WREPL_REPLICATION;
 927
 928         repl_out->command = state->command;
 929
 930         status = wreplsrv_fill_wrepl_table(service, state, table_out,
 931                                            service->wins_db->local_owner, state->full_table);
 932         NT_STATUS_NOT_OK_RETURN(status);
 933
 934         /* we won't get a reply to a inform message */
 935         state->ctrl.send_only           = True;
 936
 937         state->req = wrepl_request_send(state->wreplconn->sock, req, &state->ctrl);
 938         NT_STATUS_HAVE_NO_MEMORY(state->req);
 939
 940         state->req->async.fn            = wreplsrv_push_notify_handler_req;
 941         state->req->async.private       = state;
 942
 943         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM;
 944
 945         return NT_STATUS_OK;
 946 }
 947
 948 static NTSTATUS wreplsrv_push_notify_wait_connect(struct wreplsrv_push_notify_state *state)
 949 {
 950         NTSTATUS status;
 951
 952         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 953         NT_STATUS_NOT_OK_RETURN(status);
 954
 955         switch (state->command) {
 956         case WREPL_REPL_UPDATE:
 957                 state->full_table = True;
 958                 return wreplsrv_push_notify_update(state);
 959         case WREPL_REPL_UPDATE2:
 960                 state->full_table = False;
 961                 return wreplsrv_push_notify_update(state);
 962         case WREPL_REPL_INFORM:
 963                 state->full_table = True;
 964                 return wreplsrv_push_notify_inform(state);
 965         case WREPL_REPL_INFORM2:
 966                 state->full_table = False;
 967                 return wreplsrv_push_notify_inform(state);
 968         default:
 969                 return NT_STATUS_INTERNAL_ERROR;
 970         }
 971
 972         return NT_STATUS_INTERNAL_ERROR;
 973 }
 974
 975 static NTSTATUS wreplsrv_push_notify_wait_inform(struct wreplsrv_push_notify_state *state)
 976 {
 977         NTSTATUS status;
 978
 979         status =  wrepl_request_recv(state->req, state, NULL);
 980         NT_STATUS_NOT_OK_RETURN(status);
 981
 982         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
 983         return status;
 984 }
 985
 986 static void wreplsrv_push_notify_handler(struct wreplsrv_push_notify_state *state)
 987 {
 988         struct composite_context *c = state->c;
 989
 990         switch (state->stage) {
 991         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
 992                 c->status = wreplsrv_push_notify_wait_connect(state);
 993                 break;
 994         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM:
 995                 c->status = wreplsrv_push_notify_wait_inform(state);
 996                 break;
 997         case WREPLSRV_PUSH_NOTIFY_STAGE_DONE:
 998                 c->status = NT_STATUS_INTERNAL_ERROR;
 999         }
1000
1001         if (state->stage == WREPLSRV_PUSH_NOTIFY_STAGE_DONE) {
1002                 c->state  = COMPOSITE_STATE_DONE;
1003         }
1004
1005         if (!NT_STATUS_IS_OK(c->status)) {
1006                 c->state = COMPOSITE_STATE_ERROR;
1007         }
1008
1009         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
1010                 c->async.fn(c);
1011         }
1012 }
1013
1014 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq)
1015 {
1016         struct wreplsrv_push_notify_state *state = talloc_get_type(creq->async.private_data,
1017                                                    struct wreplsrv_push_notify_state);
1018         wreplsrv_push_notify_handler(state);
1019         return;
1020 }
1021
1022 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
1023 {
1024         struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private,
1025                                                    struct wreplsrv_push_notify_state);
1026         wreplsrv_push_notify_handler(state);
1027         return;
1028 }
1029
1030 struct composite_context *wreplsrv_push_notify_send(TALLOC_CTX *mem_ctx, struct wreplsrv_push_notify_io *io)
1031 {
1032         struct composite_context *c = NULL;
1033         struct wreplsrv_service *service = io->in.partner->service;
1034         struct wreplsrv_push_notify_state *state = NULL;
1035         enum winsrepl_partner_type partner_type;
1036
1037         c = talloc_zero(mem_ctx, struct composite_context);
1038         if (!c) goto failed;
1039
1040         state = talloc_zero(c, struct wreplsrv_push_notify_state);
1041         if (!state) goto failed;
1042         state->c        = c;
1043         state->io       = io;
1044
1045         if (io->in.inform) {
1046                 /* we can cache the connection in partner->push->wreplconn */
1047                 partner_type = WINSREPL_PARTNER_PUSH;
1048                 if (io->in.propagate) {
1049                         state->command  = WREPL_REPL_INFORM2;
1050                 } else {
1051                         state->command  = WREPL_REPL_INFORM;
1052                 }
1053         } else {
1054                 /* we can NOT cache the connection */
1055                 partner_type = WINSREPL_PARTNER_NONE;
1056                 if (io->in.propagate) {
1057                         state->command  = WREPL_REPL_UPDATE2;
1058                 } else {
1059                         state->command  = WREPL_REPL_UPDATE;
1060                 }
1061         }
1062
1063         c->state        = COMPOSITE_STATE_IN_PROGRESS;
1064         c->event_ctx    = service->task->event_ctx;
1065         c->private_data = state;
1066
1067         state->stage    = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT;
1068         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, NULL);
1069         if (!state->creq) goto failed;
1070
1071         state->creq->async.fn           = wreplsrv_push_notify_handler_creq;
1072         state->creq->async.private_data = state;
1073
1074         return c;
1075 failed:
1076         talloc_free(c);
1077         return NULL;
1078 }
1079
1080 NTSTATUS wreplsrv_push_notify_recv(struct composite_context *c)
1081 {
1082         NTSTATUS status;
1083
1084         status = composite_wait(c);
1085
1086         talloc_free(c);
1087         return status;
1088 }